Initial commit: sap-intercompany-worker

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>
This commit is contained in:
GeorgeandClaude Sonnet 5 committed 2026-09-16 07:45:57 -05:00
commit ca57403663
36 files changed
+5539

No files matched your search

+8
View File
@@ -0,0 +1,8 @@
DATABASE_URL="postgresql://intercompany:intercompany@localhost:5432/sap_intercompany?schema=public"
# Cada cuántos segundos el worker revisa eventos pendientes
POLL_INTERVAL_SECONDS=30
# Las bases SAP (incluida la casa matriz) y los escenarios de replicación ya
# NO se configuran acá: viven en Postgres. Ver scripts/manage-config.ts
# (database:add, scenario:add, etc.) y docs/ESCENARIOS.md para la guía.
+5
View File
@@ -0,0 +1,5 @@
node_modules
# Keep environment variables out of version control
.env
/src/generated/prisma
+117
View File
@@ -0,0 +1,117 @@
# sap-intercompany-worker
Worker que replica datos maestros entre bases de SAP Business One (Service
Layer), a partir de **escenarios** configurables — no hay una casa matriz
fija ni una lista fija de subsidiarias: cualquier base registrada puede ser
origen o destino, según lo que declare cada escenario.
## Qué hace
Cada `POLL_INTERVAL_SECONDS` (default 30s), revisa la tabla `webhook_events`
en Postgres — la misma que llena [`sap-webhook-api`](../sap-webhook-api) al
recibir un webhook de SAP — buscando eventos pendientes. Por cada uno:
1. Busca los **escenarios activos** que aplican a ese `businessObject`
(puede haber más de uno, ej. "clientes" y "proveedores" sobre
`BusinessPartners`, cada uno con su propio filtro), y descarta los que
no vinieron realmente de la base origen que ese escenario declara (ver
protección contra loops en `docs/ARQUITECTURA.md`).
2. Lee el registro completo desde la base origen y lo pasa por el `filter`
del escenario (`src/scenarios/<code>.ts`) — si no aplica, sigue con el
próximo. Si aplica, por cada destino lo pasa por `transform` (con acceso
a mapeos de valor cargados en Postgres vía `ctx.mapValue`) y resuelve si
ya existe en ese destino — por clave directa o por un UDF de
correspondencia, según el `keyStrategy` del escenario — para decidir
`PATCH` o `POST`.
3. Registra cada intento (escenario, verbo usado, HTTP status, error, y el
**payload ya transformado enviado**) en `replication_attempts`.
4. Calcula el estado final del evento sobre la suma de todos los intentos
de todos los escenarios que aplicaron: `done` / `partial` / `failed` /
`filtered` (ningún escenario aceptó el mensaje).
Es un proceso separado de `sap-webhook-api` a propósito: SAP tiene un
timeout de 10s para la respuesta del webhook, y replicar hacia varios
destinos (con sus propios logins y llamadas) puede tardar más que eso — por
eso el trabajo pesado corre acá, a su propio ritmo, sin poner en riesgo la
suscripción del webhook.
Ver [`docs/ARQUITECTURA.md`](./docs/ARQUITECTURA.md) para el detalle
completo del diseño (por qué escenarios y no reglas fijas, `keyStrategy`,
diagrama de flujo, reglas de agregación de estado),
[`docs/ESCENARIOS.md`](./docs/ESCENARIOS.md) para la guía paso a paso de
cómo configurar uno con ejemplos reales, y
[`docs/FLUJO.md`](./docs/FLUJO.md) para el diagrama de secuencia
archivo-por-archivo, función por función, de todo el camino de un mensaje
(incluye `sap-webhook-api`).
**Implementado**: datos maestros con destino fijo (`destinationMode = 'fixed'`
— `BP_PROVEEDORES`, `CHARTOFACCOUNTS`, `PROFITCENTERS`, `SALESPERSONS`) y
transacciones con destino resuelto desde el propio mensaje
(`destinationMode = 'payload'` — `JOURNALENTRIES`, primer caso real). Un
escenario también puede declarar un `destinationBusinessObject` distinto al
de origen (ej. `PurchaseOrders` → `Orders`), para cuando el objeto cambia
de nombre entre bases.
**Trabajo futuro**: `sap-intercompany-viewer` (frontend) +
`sap-intercompany-api` (backend nuevo, dueño de `databases`/`scenarios`/
`scenario_destinations`/`scenario_value_mappings`/`webhook_events`/
`replication_attempts` vía HTTP, ya que este worker no expone HTTP a
propósito) — visor único de eventos/log/escenarios con reproceso, en
reemplazo de `sap-webhook-viewer`.
## Quick start
```bash
npm install
cp .env.example .env # completar DATABASE_URL / POLL_INTERVAL_SECONDS
# Postgres: instancia nativa (Homebrew) en localhost:5432 — ya no se usa
# Docker para esto (docker-compose.yml queda sin uso; si tenés un Postgres
# corriendo por Homebrew/apt en el mismo puerto, ojo con el conflicto:
# localhost prioriza la instancia nativa sobre un contenedor Docker).
npx prisma migrate deploy
# Registrar bases y al menos un escenario antes de correr el worker
# (ver docs/ESCENARIOS.md para la guía completa con ejemplos reales)
npx tsx scripts/manage-config.ts database:add HQ01 "Casa Matriz" <baseUrl> <companyDb> <user> <pass>
npx tsx scripts/manage-config.ts scenario:add <code> <name> <businessObject> HQ01 <keyField>
npx tsx scripts/manage-config.ts scenario:add-destination <code> <databaseCode>
# editar src/scenarios/<code>.ts (se generó solo con la primera línea)
npm run run-once # una sola pasada, útil para probar
npm run start # corre en loop
```
Sin destinos configurados para un escenario, el worker igual corre: lee el
registro de la base origen y deja el evento en `failed` (0 intentos), sin
romper — útil para probar el flujo antes de tener todas las bases reales.
## Estructura
```
prisma/schema.prisma # modelos (ver docs/TABLAS.md)
docs/
ARQUITECTURA.md # conceptos, keyStrategy, diagrama de flujo, reglas de estado
TABLAS.md # referencia de columnas por tabla, con ejemplos reales
ESCENARIOS.md # guía paso a paso con 5 ejemplos reales configurados
FLUJO.md # diagrama de secuencia + funciones principales por archivo
scripts/manage-config.ts # CLI de administración (databases, scenarios, destinos, mapeos)
src/
config.ts # solo POLL_INTERVAL_SECONDS; todo lo demás vive en Postgres
serviceLayer.ts # cliente mínimo del Service Layer (Login/GET/POST/PATCH)
replicationConfig.ts # carga de config activa desde Postgres en cada tick
scenarioTransform.ts # carga/ejecuta filter+transform de un escenario + ctx.mapValue
replicateEvents.ts # motor: resuelve N escenarios por evento (datos maestros y transacciones)
scenarios/ # un archivo <code>.ts por escenario (filter + transform)
index.ts # cron / --once
```
## Stack
- Script plano en TypeScript (ESM), ejecutado con `tsx` — sin framework,
porque no expone HTTP.
- Prisma 7 + `@prisma/adapter-pg` contra Postgres.
- `node-cron` para el loop periódico.
Contexto completo de por qué se armó así (decisiones de arquitectura,
alternativas descartadas) en `CLAUDE.md`, en la raíz del repo.
+218
View File
@@ -0,0 +1,218 @@
# Arquitectura: motor de escenarios
## Por qué existe
La primera versión de este worker tenía una regla fija por cada `businessObject`
recibido del webhook (una sola por objeto, aplicada siempre, hacia todas las
subsidiarias activas). Eso alcanzaba mientras solo existía **un** caso de uso
por objeto. En cuanto aparece más de una forma de tratar el mismo objeto —
por ejemplo, replicar los **clientes** de `BusinessPartners` con reglas y
destinos distintos que los **proveedores** — la regla 1:1 no alcanza.
El motor actual reemplaza esa regla única por **escenarios**: cada uno es una
forma completa e independiente de procesar un `businessObject`, con su
propio filtro, base origen, base(s) destino y transformación de payload.
Puede haber varios escenarios activos para el mismo objeto a la vez, y un
mismo evento puede ser procesado por **más de uno** en paralelo.
## Conceptos
### Base de datos (`databases`)
Registro único de toda compañía SAP alcanzable por Service Layer: URL,
`CompanyDB` y credenciales. **No existe distinción entre "casa matriz" y
"subsidiaria"** — cualquier fila puede ser origen en un escenario y destino
en otro al mismo tiempo. Ver [`TABLAS.md`](./TABLAS.md#databases).
### Escenario (`scenarios`)
La unidad central de configuración. Cada escenario define:
- a qué `businessObject` aplica (`BusinessPartners`, `SalesPersons`, ...) —
también determina con qué se leen los registros del origen,
- opcionalmente, un `destinationBusinessObject` distinto — si no se
especifica, el destino usa el mismo objeto que el origen (el caso
común); se especifica cuando el objeto cambia de nombre entre bases
(ej. origen `PurchaseOrders` → destino `Orders`, comercio intercompañía),
- de qué base lee el registro completo (`originDatabaseCode`),
- qué campo identifica al registro (`keyField`, ej. `CardCode`) y de qué
tipo es (`keyType`: `string` — va entre comillas en la URL OData — o
`number` — va sin comillas; mandar comillas en un campo numérico da
HTTP 400),
- cómo se resuelve si el registro ya existe en el destino (`keyStrategy`,
ver más abajo),
- el modo de resolver destino: `fixed` (lista fija de bases en
`scenario_destinations`, para datos maestros) o `payload` (el destino se
resuelve leyendo un `routingField` del propio mensaje, para
transacciones) — **los dos modos están implementados**, ver el ejemplo
real de `JOURNALENTRIES` en [`ESCENARIOS.md`](./ESCENARIOS.md).
El filtro (si el escenario aplica a un registro puntual) y la
transformación del payload NO son columnas de esta tabla — son las dos
partes fijas del archivo de código de cada escenario, ver más abajo.
Ver [`ESCENARIOS.md`](./ESCENARIOS.md) para la guía completa de cómo armar
uno paso a paso, con ejemplos reales.
### `keyStrategy`: cómo se identifica el mismo registro entre bases
- **`portable`**: el valor de `keyField` significa lo mismo en origen y
destino (`CardCode`, `CenterCode` — códigos de negocio asignados a mano).
El motor hace `GET` directo por esa clave en el destino.
- **`udfLink`**: SAP asigna la clave por base (ej. `SalesEmployeeCode`, que
autonumera cada compañía) — el mismo vendedor tiene un código distinto en
cada subsidiaria. El escenario declara un `matchField`: un UDF creado a
mano en el destino (ej. `U_HQ_SlpCode`) que guarda el valor de la clave de
origen. El motor busca por ese UDF (`GET .../SalesPersons?$filter=U_HQ_SlpCode eq '5'`)
en vez de por la clave directa:
- si encuentra un registro, usa SU clave propia (la del destino) para el
`PATCH`,
- si no encuentra nada, hace `POST` (con el UDF seteado en el payload
para poder encontrarlo la próxima vez) y guarda la clave que SAP le
asignó en el destino como `resultKey` del intento, para trazabilidad.
### Destinos de un escenario (`scenario_destinations`)
Solo aplica a escenarios `destinationMode = 'fixed'`. Es la lista de bases
hacia las que ese escenario replica. Un escenario puede tener una o varias.
### El archivo del escenario: `src/scenarios/<code>.ts`
Cada escenario tiene **un solo archivo**, con **siempre las mismas dos
partes**:
```ts
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean | Promise<boolean>
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown>
```
**`filter`** decide si el escenario aplica a ESE registro puntual (`true` =
aplica). Vive en código — no en columnas de la tabla `scenarios` — porque
una condición simple (campo/operador/valor) no alcanza para casos reales
(condiciones compuestas, varios campos a la vez). Se evalúa una sola vez
por evento, contra el registro leído de la base origen, y puede ser
`async`: recibe `ctx.get(path)`, un `GET` arbitrario contra la base ORIGEN
del escenario (reusa la sesión ya autenticada, no abre un login nuevo).
Esto cubre el caso típico de transacciones: una Factura trae el `CardCode`
del cliente en su payload, pero el filtro depende de un campo del
`BusinessPartner` (ej. su `GroupCode`) — el archivo hace
`await ctx.get("/BusinessPartners('${record.CardCode}')")` y decide con
ese dato, sin que el motor tenga que saber nada de esa relación.
**`transform`** arma el payload a enviar a un destino puntual — se llama
**una vez por destino**, recibe el registro ya leído (limpio de metadata
OData) y un `ctx` con:
- `ctx.destinationCode`: a qué base le toca este payload (permite
transformar distinto según el destino).
- `ctx.mapValue(fieldName, sourceValue)`: consulta la tabla
`scenario_value_mappings` (ver [`TABLAS.md`](./TABLAS.md)) para ESTE
escenario y ESTE destino. Devuelve `undefined` si no hay mapeo cargado —
el archivo decide qué hacer (tirar error para marcar el intento como
`SKIPPED`, dejar el campo afuera, poner un default, etc.). Esta es la
parte de **mapeo** dentro de la transformación.
Filtro y transformación viven en código porque necesitan lógica arbitraria
(condicionales, elegir campos) que no es expresable como una tabla de
equivalencias simple — pero los VALORES de los mapeos de campo (ej.
`GroupCode` de HQ → `GroupCode` de cada subsidiaria) sí son datos puros, y
esos viven en Postgres para poder cargarlos con el CLI sin tocar código (y
a futuro, desde una UI). El archivo sigue siendo quien decide QUÉ campo
mapear y QUÉ hacer si falta el valor — la tabla solo guarda el dato. Ver
[`ESCENARIOS.md`](./ESCENARIOS.md) para los 3 ejemplos completos ya
configurados.
### Intento de replicación (`replication_attempts`)
Un registro por cada intento de un escenario hacia una base destino
puntual: verbo usado, HTTP status, mensaje de error y el **payload ya
transformado que se envió** (siempre, en éxito y en falla — pensado para
una futura UI de consulta, ver "Trabajo futuro" más abajo). Es un log de
auditoría inmutable: guarda los códigos de base/escenario como texto en el
momento del intento, no relaciones vivas, para no perder trazabilidad si la
configuración cambia o se borra después.
## Flujo de un evento
Diagrama de decisión (qué pasa y por qué). Para el diagrama de secuencia
archivo-por-archivo, función por función — incluyendo `sap-webhook-api` —
ver [`FLUJO.md`](./FLUJO.md).
```mermaid
flowchart TD
A[webhook_events: status=pending] --> B[Tomar evento]
B --> C{Por cada escenario activo<br/>del mismo businessObject}
C --> S{source del evento coincide<br/>con la base origen del escenario?}
S -- no --> C
S -- si --> D[Leer registro completo<br/>desde la base origen del escenario]
D --> E{filter-record- del<br/>archivo del escenario}
E -- false --> C
E -- true --> F[transform-record, ctx-<br/>por cada base destino]
F --> K{keyStrategy}
K -- portable --> G1[GET directo por keyField]
K -- udfLink --> G2[GET por filtro sobre matchField]
G1 --> G[Existe en el destino?]
G2 --> G
G -- si --> H[PATCH]
G -- no --> I[POST + capturar clave asignada]
H --> J[Registrar intento<br/>verbo, status, error, payload]
I --> J
J --> C
C -- todos los escenarios procesados --> L[Calcular status final del evento]
```
### Protección contra loops: `source` del evento vs. base origen
El CloudEvent que manda SAP incluye `"source": "/default/sap.b1/<CompanyDB>"`
— de qué base salió la notificación. Antes de procesar un evento con un
escenario, el motor compara ese `CompanyDB` contra el de la base origen
configurada (`originDatabase.companyDb`). Si no coincide, el escenario
simplemente no aplica (no genera ningún intento).
Esto no es cosmético: se descubrió en producción-de-prueba que si una base
DESTINO tiene su propia suscripción de webhook activa (ej. porque se creó
clonando/restaurando la base origen y heredó su configuración), un `PATCH`
nuestro ahí dispara una notificación que, sin este chequeo, el motor
reprocesaría como si fuera un cambio nuevo en el origen — generando un
`PATCH` de nuevo, que dispara otra notificación, indefinidamente. El fix de
fondo es desactivar `EnableWebhook` en las bases que son solo destino, pero
este chequeo queda como segunda capa de protección barata (una comparación
de string) para cualquier otro caso de mala configuración futura.
### Estado final del evento (`webhook_events.status`)
Como un evento puede ser tomado por varios escenarios a la vez, el estado
final se calcula sobre **todos** los intentos de **todos** los escenarios
que aplicaron:
| Estado | Condición |
|---|---|
| `filtered` | Ningún escenario activo para ese `businessObject` aceptó el mensaje (ni siquiera vino de la base origen esperada de ninguno, o todos los filtros lo rechazaron). |
| `done` | Al menos un escenario aplicó, y **todos** los intentos (de todos los escenarios, hacia todos sus destinos) tuvieron éxito. |
| `partial` | Al menos un escenario aplicó, y hay una mezcla de intentos exitosos y fallidos. |
| `failed` | Al menos un escenario aplicó, pero **ningún** intento tuvo éxito (incluye el caso de un escenario sin destinos configurados). |
## `destinationMode = 'payload'` (transacciones) — implementado
`processPendingEvents` (en `src/replicateEvents.ts`) procesa ambos modos. En `payload`, no hay filas
en `scenario_destinations` para ese escenario — el destino se resuelve en
runtime leyendo `sourceRecord[scenario.routingField]` (un UDF en el
**origen** cuyo valor es directamente el `code` de la base destino, ej.
`SUB_01` — no el nombre de la compañía SAP). Si el campo viene vacío o no
matchea ninguna base registrada, se registra un intento de error (destino
`'N/A'`) en vez de fallar en silencio. Ejemplo real funcionando:
`JOURNALENTRIES` (ver [`ESCENARIOS.md`](./ESCENARIOS.md) sección 5).
`keyStrategy=udfLink` es especialmente relevante en modo `payload`: como
cada mensaje puede terminar en una base distinta, sin un UDF de
correspondencia (`matchField`) reprocesar el mismo mensaje generaría un
documento duplicado en el destino cada vez.
## Trabajo futuro: `sap-intercompany-viewer`
Proyecto aparte (a construir después de terminar este) para la parte visual
de toda esta configuración: visor de eventos y reenvíos, alta de bases de
datos con prueba de conexión, alta/edición de escenarios, y administración
de sus mapeos de valor. `replication_attempts` ya guarda siempre el payload
enviado (éxito y falla) e incluye índices sobre `event_id`, `scenario_id` y
`attempted_at` pensando en ese consumo futuro.
+259
View File
@@ -0,0 +1,259 @@
# Guía: cómo armar un escenario
Toda la configuración se administra con `scripts/manage-config.ts` (vía
`tsx`), salvo el filtro y la transformación de cada escenario, que son
código. Los cambios tienen efecto en el próximo tick del worker — no hace
falta reiniciar el proceso, tampoco después de editar el archivo del
escenario: el motor lo importa dinámicamente en cada intento.
**Nombre del archivo**: el `code` de un escenario va en mayúsculas (es el
identificador que ves en `scenario:list`, en los logs y en `--filter`/etc.)
pero su archivo en `src/scenarios/` va en minúsculas — ej. código
`BP_PROVEEDORES` → archivo `bp_proveedores.ts`. `scenario:add` ya genera el
archivo con el nombre correcto; si necesitás renombrar un escenario
después, usá `scenario:rename` (mueve el archivo automáticamente, no lo
hagas a mano).
## 1. Registrar las bases de datos
Antes de crear un escenario, las bases que va a usar (origen y destinos)
tienen que existir en `databases`:
```bash
npx tsx scripts/manage-config.ts database:add HQ01 "HQ DemoCL" https://sap10:50000/b1s/v2 SBODemoCL manager mngr
npx tsx scripts/manage-config.ts database:add SUB_01 "SUB 01 DemoCL" https://sap10:50000/b1s/v2 SUBDemoCL_01 manager mngr
npx tsx scripts/manage-config.ts database:add SUB_02 "SUB 02 DemoCL" https://sap10:50000/b1s/v2 SUBDemoCL_02 manager mngr
npx tsx scripts/manage-config.ts database:list
```
**Importante**: si una base destino se creó clonando/restaurando la base
origen (común en ambientes de demo), puede haber heredado su propia
suscripción de webhook activa. Si el mismo `businessObject` que estás por
replicar dispara un webhook en el destino, hay que desactivarlo
(`EnableWebhook: tNO` en esa compañía) — si no, cada `PATCH`/`POST` nuestro
genera una notificación que, aunque el motor la descarta (ver
`ARQUITECTURA.md` § protección contra loops), sigue entrando como ruido.
## 2. Ejemplo 1: clave portable + mapeo de valor (`BP_PROVEEDORES`)
Caso real: `BusinessPartners` con `CardType=cSupplier`, replicando solo
algunos campos, con `GroupCode` traducido por destino.
### 2.1 Crear el escenario
```bash
npx tsx scripts/manage-config.ts scenario:add BP_PROVEEDORES "BP - Proveedores" BusinessPartners HQ01 CardCode
```
`CardCode` es texto de negocio (igual en HQ01 y en cada subsidiaria), así
que no hace falta `--key-strategy` ni `--key-type` — quedan en sus
defaults (`portable` y `string`).
Este comando genera automáticamente `src/scenarios/bp_proveedores.ts` (si
no existía) con un `filter`/`transform` de partida (`filter` devuelve
`true` siempre, `transform` devuelve el registro tal cual) — hay que
editarlo con la condición real.
### 2.2 Definir los destinos
```bash
npx tsx scripts/manage-config.ts scenario:add-destination BP_PROVEEDORES SUB_01
npx tsx scripts/manage-config.ts scenario:add-destination BP_PROVEEDORES SUB_02
```
### 2.3 Cargar los mapeos de valor (`GroupCode`)
`GroupCode` en SAP significa "grupo de clientes" o "grupo de proveedores"
según el `CardType` del registro — son dominios de valores distintos,
aunque el nombre del campo sea el mismo. Por eso el mapeo se carga **por
escenario**, no global:
```bash
npx tsx scripts/manage-config.ts scenario:mapping:add BP_PROVEEDORES GroupCode SUB_01 101 101
npx tsx scripts/manage-config.ts scenario:mapping:add BP_PROVEEDORES GroupCode SUB_02 101 101
# ...un comando por cada (destino, valor de origen) — ver docs/TABLAS.md para la lista completa
```
### 2.4 Escribir el filtro y la transformación
`src/scenarios/bp_proveedores.ts` — el archivo siempre tiene las mismas
dos partes. El filtro es la condición `CardType=cSupplier` (antes vivía
como columnas en la tabla `scenarios`, ahora es código); la transformación
selecciona solo los campos necesarios y resuelve `GroupCode` con
`ctx.mapValue`, no con el valor crudo de HQ01:
```ts
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: decide si este escenario aplica al registro ---
// No necesita ctx acá: CardType ya viene en el payload de BusinessPartners.
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean {
return record.CardType === 'cSupplier';
}
// --- Transformación: arma el payload para un destino puntual ---
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
// --- Mapeo: GroupCode vive en Postgres (scenario_value_mappings), no acá ---
const groupCode = record.GroupCode === undefined || record.GroupCode === null ? undefined : String(record.GroupCode);
const mappedGroupCode = groupCode === undefined ? undefined : ctx.mapValue('GroupCode', groupCode);
if (groupCode !== undefined && mappedGroupCode === undefined) {
throw new Error(`GroupCode '${groupCode}' no tiene mapeo cargado para el destino '${ctx.destinationCode}'`);
}
return {
CardCode: record.CardCode,
CardName: record.CardName,
CardType: record.CardType,
GroupCode: mappedGroupCode,
Phone1: record.Phone1,
Phone2: record.Phone2,
Notes: record.Notes,
FederalTaxID: record.FederalTaxID,
Currency: record.Currency,
DebitorAccount: record.DebitorAccount,
Valid: record.Valid,
Frozen: record.Frozen,
Properties1: record.Properties1,
};
}
```
Si `filter` devuelve `false`, el escenario no aplica a ese registro (no
genera ningún intento). Si `GroupCode` viene con un valor sin mapeo
cargado dentro de `transform`, el `throw` marca el intento como `SKIPPED`
(no manda nada sin traducir) y el motivo queda en `error_message` de
`replication_attempts`.
## 3. Ejemplo 2: clave asignada por base (`SALESPERSONS`, `keyStrategy=udfLink`)
`SalesPersons` usa `SalesEmployeeCode`, que SAP autonumera **por
compañía** — el mismo vendedor tiene un código distinto en cada
subsidiaria. No se puede usar `portable`.
### 3.1 Crear el UDF en SAP (paso manual, fuera de este repo)
En cada subsidiaria destino, crear un campo de usuario en la tabla `OSLP`
(Sales Employees) — ej. `U_HQ_SlpCode`, tipo **Alfanumérico** — desde
Herramientas de Personalización / UDF Manager de SAP B1. Sin esto, el
escenario no tiene dónde guardar la correspondencia.
### 3.2 Crear el escenario
```bash
npx tsx scripts/manage-config.ts scenario:add SALESPERSONS "SalesPersons" SalesPersons HQ01 SalesEmployeeCode \
--key-strategy udfLink --match-field U_HQ_SlpCode --key-type number
npx tsx scripts/manage-config.ts scenario:add-destination SALESPERSONS SUB_01
npx tsx scripts/manage-config.ts scenario:add-destination SALESPERSONS SUB_03
```
`--key-type number` es necesario porque `SalesEmployeeCode` es numérico en
el Service Layer — mandar la URL con comillas (`SalesPersons('5')`) da
HTTP 400; sin comillas (`SalesPersons(5)`) funciona.
### 3.3 Cómo lo resuelve el motor (sin que tengas que hacer nada más)
- Si ya existe un `SalesPersons` en el destino con `U_HQ_SlpCode` igual al
`SalesEmployeeCode` de origen → lo encuentra, usa la clave propia del
destino para el `PATCH`.
- Si no existe ninguno → `POST` (con `U_HQ_SlpCode` seteado en el payload
automáticamente por el motor — no hace falta ponerlo en el `transform`),
y guarda la clave que SAP le asignó en `replication_attempts.result_key`
para la próxima actualización.
Sin lógica de negocio adicional pedida para este objeto, el `filter`
(`return true`) y el `transform` (identity, `return { ...record }`)
generados por `scenario:add` alcanzan tal cual.
## 4. Ejemplo 3: sin filtro, a 3 destinos (`PROFITCENTERS`)
`ProfitCenters` usa `CenterCode`, un código de negocio asignado a mano
(`CC1`, `Stelle_2`) — `portable`, igual que `BusinessPartners`:
```bash
npx tsx scripts/manage-config.ts scenario:add PROFITCENTERS "ProfitCenters" ProfitCenters HQ01 CenterCode
npx tsx scripts/manage-config.ts scenario:add-destination PROFITCENTERS SUB_01
npx tsx scripts/manage-config.ts scenario:add-destination PROFITCENTERS SUB_02
npx tsx scripts/manage-config.ts scenario:add-destination PROFITCENTERS SUB_03
```
Sin condición pedida: `filter` queda como lo genera el scaffold
(`return true`), aplica a todo `ProfitCenters`.
## 5. Filtro que necesita un dato que no viene en el payload
Caso típico de transacciones: una `Invoices` (Factura de Venta) trae el
`CardCode` del cliente, pero el escenario solo debe aplicar si ese cliente
pertenece al `GroupCode` 101. El dato para decidir el filtro no está en el
payload de la Factura — hay que ir a buscarlo a `BusinessPartners`. Por
eso `filter` puede ser `async` y usar `ctx.get(...)` (mismo cliente/sesión
que ya trae el registro principal, sin loguearse de nuevo):
```ts
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: necesita el GroupCode del cliente, que no viene en la Factura ---
export async function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): Promise<boolean> {
const cardCode = record.CardCode;
if (typeof cardCode !== 'string') return false;
const res = await ctx.get(`/BusinessPartners('${cardCode}')`);
if (res.status !== 200) {
throw new Error(`No se pudo leer BusinessPartners('${cardCode}') de ${ctx.originCode}: HTTP ${res.status}`);
}
return res.data.GroupCode === 101;
}
```
Si `ctx.get` tira o devuelve un status inesperado, el `throw` marca el
intento como error (destino `'N/A'`, ver `ARQUITECTURA.md`) — no rompe la
corrida completa, solo ese evento en ese escenario.
**Importante**: esto resuelve la parte del filtro enriquecido, pero
`Invoices` es una transacción (`destinationMode = 'payload'`), y ese modo
**todavía no está implementado** en el motor (ver limitación en
`ARQUITECTURA.md`) — hace falta además el ruteo del destino desde el
propio mensaje antes de poder usar este escenario de punta a punta.
## 6. Comandos disponibles (referencia)
```
database:add <code> <name> <baseUrl> <companyDb> <user> <pass>
database:disable <code>
database:list
scenario:add <code> <name> <businessObject> <originDbCode> <keyField>
[--key-strategy udfLink --match-field U_HQ_SlpCode] [--key-type number]
[--destination-mode payload --routing-field U_ITC_Target]
[--destination-object Orders]
scenario:rename <oldCode> <newCode>
scenario:disable <code>
scenario:list
scenario:add-destination <scenarioCode> <databaseCode>
scenario:remove-destination <scenarioCode> <databaseCode>
scenario:mapping:add <scenarioCode> <fieldName> <databaseCode> <sourceValue> <targetValue>
scenario:mapping:list <scenarioCode> <fieldName>
```
`--destination-object` es para cuando el objeto cambia de nombre entre
origen y destino — ej. un escenario que lee `PurchaseOrders` de HQ01 pero
necesita crear un `Orders` (Orden de Venta) en la subsidiaria, típico de
comercio intercompañía. Si se omite, el destino usa el mismo objeto que
`<businessObject>` (el caso de los 5 escenarios de hoy).
`scenario:add` y `database:add` son upsert por `code`: correr el mismo
comando de nuevo actualiza la fila existente en vez de duplicarla. La
primera vez que se crea un escenario nuevo, además genera su archivo en
`src/scenarios/` con un `filter`/`transform` de partida — no lo sobrescribe
en corridas siguientes. `scenario:rename` cambia el `code` en la DB **y**
mueve el archivo correspondiente en un solo paso — no renombrar el archivo
a mano, porque el motor lo busca por el `code` actual. `scenario:disable` /
`database:disable` desactivan sin borrar — útil para pausar un escenario o
sacar una base de servicio sin perder su configuración ni el historial de
`replication_attempts` que la referencia.
+134
View File
@@ -0,0 +1,134 @@
# 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`.
Binary file not shown.
+156
View File
@@ -0,0 +1,156 @@
# Referencia de tablas
Todas viven en la base Postgres compartida (`docker-compose.yml` en la raíz
del workspace), definidas en [`prisma/schema.prisma`](../prisma/schema.prisma).
Los ejemplos de esta página usan datos reales de la config actual del
ambiente de desarrollo (bases `HQ01`/`SUB_01`.."SUB_04", escenario
`BP_PROVEEDORES`).
## `webhook_events`
Bandeja de entrada. `sap-webhook-api` inserta una fila acá por cada
notificación real que recibe de SAP (además de guardar el `.json` local).
No se toca a mano — el worker es el único que la actualiza.
| Columna | Tipo | Notas |
|---|---|---|
| `id` | `String` (PK) | El `id` del CloudEvent que manda SAP. |
| `business_object` | `String` | `'BusinessPartners'`, `'SalesPersons'`, `'ProfitCenters'`, ... |
| `operation` | `String` | `'Created'` \| `'Updated'`. |
| `subject` | `String` | Clave del registro en la base que lo generó (`CardCode`, `SalesEmployeeCode`, `CenterCode`, según el objeto). |
| `payload` | `Json` | El CloudEvent completo tal cual llegó — incluye `source` (`/default/sap.b1/<CompanyDB>`), que el motor usa para la protección contra loops (ver `ARQUITECTURA.md`). |
| `status` | `EventStatus` | `pending` → `processing` → `done` \| `partial` \| `failed` \| `filtered`. |
| `received_at` | `DateTime` | |
## `databases`
Registro único de toda base SAP alcanzable por Service Layer. Sin
distinción HQ/subsidiaria — ver [`ARQUITECTURA.md`](./ARQUITECTURA.md).
| Columna | Tipo | Notas |
|---|---|---|
| `id` | `Int` (PK, autoincremental) | Identificador interno, no se usa como referencia legible. |
| `code` | `String` (único) | Mnemónico corto y **estable**, asignado una sola vez al crear (ej. `HQ01`, `SUB_01`). No autoincremental — es lo que aparece en los logs. |
| `name` | `String` | Descriptivo, solo para mostrar. |
| `base_url` | `String` | URL del Service Layer, ej. `https://sap10:50000/b1s/v2`. |
| `company_db` | `String` | `CompanyDB` de esa base. |
| `username` / `password` | `String` | Credenciales de login del Service Layer. |
| `active` | `Boolean` | Una base inactiva no se usa como origen ni destino en ningún escenario. |
Ejemplo (ambiente de desarrollo actual):
| code | name | base_url | company_db | active |
|---|---|---|---|---|
| `HQ01` | HQ DemoCL | `https://sap10:50000/b1s/v2` | `SBODemoCL` | `true` |
| `SUB_01` | SUB 01 DemoCL | `https://sap10:50000/b1s/v2` | `SUBDemoCL_01` | `true` |
| `SUB_02` | SUB 02 DemoCL | `https://sap10:50000/b1s/v2` | `SUBDemoCL_02` | `true` |
| `SUB_03` | SUB 03 DemoCL | `https://sap10:50000/b1s/v2` | `SUBDemoCL_03` | `true` |
| `SUB_04` | SUB 04 DemoCL | `https://sap10:50000/b1s/v2` | `SUBDemoCL_04` | `true` |
## `scenarios`
Una fila por escenario. Ver [`ESCENARIOS.md`](./ESCENARIOS.md) para cómo
armar uno con el CLI.
| Columna | Tipo | Notas |
|---|---|---|
| `id` | `Int` (PK) | |
| `code` | `String` (único) | Identificador legible del escenario, en mayúsculas (ej. `BP_PROVEEDORES`). Determina el nombre de su archivo (`filter` + `transform`) en minúsculas: `src/scenarios/<code en minúsculas>.ts` (ej. `bp_proveedores.ts`) — ver `scenarioFileSlug` en `src/scenarioTransform.ts`. |
| `name` | `String` | Descriptivo. |
| `business_object` | `String` | Objeto que dispara el escenario: con qué se matchea `webhook_events` y con qué se lee el registro del origen. |
| `destination_business_object` | `String?` | Objeto a usar en el `GET`/`POST`/`PATCH` del **destino**, si es distinto al de origen (ej. origen `PurchaseOrders` → destino `Orders`, típico de comercio intercompañía). `null` ⇒ mismo objeto que `business_object` (el caso común). |
| `origin_database_code` | `String` (FK → `databases.code`) | De dónde se lee el registro completo. |
| `destination_mode` | `ScenarioDestinationMode` | `fixed` (lista fija en `scenario_destinations`) \| `payload` (destino resuelto por `routing_field` — ambos implementados). |
| `key_field` | `String` | Campo clave del registro, ej. `CardCode`, `SalesEmployeeCode`. |
| `key_type` | `ScenarioKeyType` | `string` (va entre comillas en la URL OData) \| `number` (sin comillas — un campo numérico con comillas da HTTP 400). |
| `key_strategy` | `ScenarioKeyStrategy` | `portable` (la clave vale lo mismo en origen y destino) \| `udfLink` (SAP asigna la clave por base; se busca por `match_field`). |
| `match_field` | `String?` | Solo si `key_strategy = udfLink`: nombre del UDF en el destino que guarda la clave de origen (ej. `U_ITC_Source_JdtNum`). |
| `routing_field` | `String?` | Solo si `destination_mode = payload`: UDF en el origen cuyo valor es el `code` de la base destino (ej. `U_ITC_Target`). |
| `active` | `Boolean` | |
No hay columnas de filtro: el filtro es la función `filter()` del archivo
del escenario (ver [`ARQUITECTURA.md`](./ARQUITECTURA.md)).
Ejemplo (los 5 escenarios activos hoy en desarrollo):
| code | business_object | destination_business_object | origin | destinations | key_field | key_strategy |
|---|---|---|---|---|---|---|
| `BP_PROVEEDORES` | `BusinessPartners` | — | `HQ01` | `SUB_01`, `SUB_02` | `CardCode` | `portable` |
| `CHARTOFACCOUNTS` | `ChartOfAccounts` | — | `HQ01` | `SUB_01`, `SUB_02` | `Code` | `portable` |
| `JOURNALENTRIES` | `JournalEntries` | — | `HQ01` | (payload: `U_ITC_Target`) | `JdtNum` | `udfLink` (`U_ITC_Source_JdtNum`) |
| `PROFITCENTERS` | `ProfitCenters` | — | `HQ01` | `SUB_01`, `SUB_02`, `SUB_03` | `CenterCode` | `portable` |
| `SALESPERSONS` | `SalesPersons` | — | `HQ01` | `SUB_01`, `SUB_03` | `SalesEmployeeCode` | `udfLink` (`U_HQ_SlpCode`) |
Ninguno usa todavía `destination_business_object` (ninguno lo necesita por
ahora) — quedaría, por ejemplo, `PurchaseOrders` → `Orders` en un futuro
escenario de compras intercompañía.
## `scenario_destinations`
Destinos fijos de un escenario (solo aplica a `destination_mode = fixed`).
Único por `(scenario_id, database_code)`.
| Columna | Tipo | Notas |
|---|---|---|
| `scenario_id` | `Int` (FK → `scenarios.id`) | |
| `database_code` | `String` (FK → `databases.code`) | |
## `scenario_value_mappings`
Datos puros que el archivo de transformación de un escenario consulta vía
`ctx.mapValue(fieldName, sourceValue)` — ver
[`ARQUITECTURA.md`](./ARQUITECTURA.md). El código decide QUÉ campo mapear y
QUÉ hacer si falta el valor (tirar error, dejar sin traducir, etc.); acá
solo vive la equivalencia. Escopada por escenario (no global por
objeto+campo) porque el mismo campo puede tener dominios de valores
distintos según el escenario (ej. `GroupCode` de clientes vs. de
proveedores).
| Columna | Tipo | Notas |
|---|---|---|
| `scenario_id` | `Int` (FK) | |
| `field_name` | `String` | Ej. `GroupCode`. |
| `database_code` | `String` (FK → `databases.code`) | El destino al que aplica esta traducción puntual. |
| `source_value` | `String` | Valor tal como viene de la base origen. |
| `target_value` | `String` | Valor a usar en ese destino. |
Único por `(scenario_id, field_name, database_code, source_value)`.
Ejemplo real (`BP_PROVEEDORES`, `GroupCode`, hoy 1:1 en ambas subsidiarias):
| scenario | field_name | database_code | source_value | target_value |
|---|---|---|---|---|
| `BP_PROVEEDORES` | `GroupCode` | `SUB_01` | `101` | `101` |
| `BP_PROVEEDORES` | `GroupCode` | `SUB_01` | `110` | `110` |
| `BP_PROVEEDORES` | `GroupCode` | `SUB_02` | `101` | `101` |
| ... | | | | (7 códigos × 2 destinos = 14 filas) |
## `replication_attempts`
Log de auditoría inmutable. Una fila por cada intento de un escenario hacia
una base destino.
| Columna | Tipo | Notas |
|---|---|---|
| `event_id` | `String` (FK → `webhook_events.id`) | |
| `scenario_id` | `Int` (FK → `scenarios.id`) | Imprescindible porque un evento puede pasar por varios escenarios a la vez. |
| `origin_database` | `String` | Código de la base origen usada en **este** intento — texto libre, no FK: sobrevive si la base se borra o renombra después. |
| `destination_database` | `String` | Ídem para el destino. `'N/A'` si el intento falló antes de resolver un destino (ej. error leyendo el origen). |
| `verb_used` | `String` | `POST` \| `PATCH` \| `SKIPPED` (falta mapeo, o transform tiró error) \| `N/A` (error antes de llamar al destino). |
| `http_status` | `Int?` | |
| `error_message` | `String?` | |
| `payload_sent` | `Json?` | El payload ya transformado enviado al destino — se guarda **siempre**, en éxito y en falla. |
| `result_key` | `String?` | Para `key_strategy=udfLink`: la clave propia del destino (encontrada o recién asignada por SAP en un `POST`). `null` en `key_strategy=portable` y en datos maestros sin esta estrategia. |
| `attempted_at` | `DateTime` | |
Índices sobre `event_id`, `scenario_id` y `attempted_at`, pensados para la
futura UI de consulta (`sap-intercompany-viewer`, ver `ARQUITECTURA.md`).
Ejemplo real (`SALESPERSONS`, vendedor nuevo `SalesEmployeeCode=7` en HQ01,
`POST` en ambas subsidiarias — SAP le asignó el mismo número por
coincidencia, pero el motor no lo asume, lo captura del response):
| event_id | scenario | destination_database | verb_used | http_status | result_key |
|---|---|---|---|---|---|
| `149be29e-...` | `SALESPERSONS` | `SUB_01` | `POST` | `201` | `7` |
| `149be29e-...` | `SALESPERSONS` | `SUB_03` | `POST` | `201` | `7` |
+2968
View File
File diff suppressed because it is too large. Load diff
+32
View File
@@ -0,0 +1,32 @@
{
"name": "sap-intercompany-worker",
"version": "1.0.0",
"description": "Replica datos maestros de SAP HeadQuarter hacia subsidiarias",
"type": "module",
"main": "src/index.ts",
"scripts": {
"start": "tsx src/index.ts",
"dev": "tsx watch src/index.ts",
"run-once": "tsx src/index.ts --once"
},
"keywords": [],
"author": "",
"license": "UNLICENSED",
"private": true,
"dependencies": {
"@prisma/adapter-pg": "^7.9.1",
"@prisma/client": "^7.9.1",
"axios": "^1.19.0",
"dotenv": "^17.4.2",
"node-cron": "^4.6.0",
"pg": "^8.23.0",
"prisma": "^7.9.1"
},
"devDependencies": {
"@types/node": "^26.2.0",
"@types/node-cron": "^3.0.11",
"@types/pg": "^8.23.1",
"tsx": "^4.23.12",
"typescript": "^7.0.2"
}
}
+14
View File
@@ -0,0 +1,14 @@
// This file was generated by Prisma, and assumes you have installed the following:
// npm install --save-dev prisma dotenv
import "dotenv/config";
import { defineConfig } from "prisma/config";
export default defineConfig({
schema: "prisma/schema.prisma",
migrations: {
path: "prisma/migrations",
},
datasource: {
url: process.env["DATABASE_URL"],
},
});
@@ -0,0 +1,31 @@
-- CreateEnum
CREATE TYPE "EventStatus" AS ENUM ('pending', 'processing', 'done', 'partial', 'failed');
-- CreateTable
CREATE TABLE "webhook_events" (
"id" TEXT NOT NULL,
"business_object" TEXT NOT NULL,
"operation" TEXT NOT NULL,
"subject" TEXT NOT NULL,
"payload" JSONB NOT NULL,
"received_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"status" "EventStatus" NOT NULL DEFAULT 'pending',
CONSTRAINT "webhook_events_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "replication_attempts" (
"id" SERIAL NOT NULL,
"event_id" TEXT NOT NULL,
"subsidiary" TEXT NOT NULL,
"verb_used" TEXT NOT NULL,
"http_status" INTEGER,
"error_message" TEXT,
"attempted_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "replication_attempts_pkey" PRIMARY KEY ("id")
);
-- AddForeignKey
ALTER TABLE "replication_attempts" ADD CONSTRAINT "replication_attempts_event_id_fkey" FOREIGN KEY ("event_id") REFERENCES "webhook_events"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
@@ -0,0 +1,83 @@
-- CreateEnum
CREATE TYPE "ReplicationRuleType" AS ENUM ('master', 'transaction');
-- AlterEnum
ALTER TYPE "EventStatus" ADD VALUE 'filtered';
-- AlterTable
ALTER TABLE "replication_attempts" ADD COLUMN "result_key" TEXT;
-- CreateTable
CREATE TABLE "subsidiaries" (
"id" SERIAL NOT NULL,
"code" TEXT NOT NULL,
"name" TEXT NOT NULL,
"base_url" TEXT NOT NULL,
"company_db" TEXT NOT NULL,
"username" TEXT NOT NULL,
"password" TEXT NOT NULL,
"routing_value" TEXT,
"active" BOOLEAN NOT NULL DEFAULT true,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "subsidiaries_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "replication_rules" (
"id" SERIAL NOT NULL,
"business_object" TEXT NOT NULL,
"type" "ReplicationRuleType" NOT NULL,
"key_field" TEXT,
"routing_field" TEXT,
"idempotency_field" TEXT,
"active" BOOLEAN NOT NULL DEFAULT true,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "replication_rules_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "mapped_fields" (
"id" SERIAL NOT NULL,
"business_object" TEXT NOT NULL,
"field_name" TEXT NOT NULL,
"active" BOOLEAN NOT NULL DEFAULT true,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "mapped_fields_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "reference_field_mappings" (
"id" SERIAL NOT NULL,
"business_object" TEXT NOT NULL,
"field_name" TEXT NOT NULL,
"subsidiary_code" TEXT NOT NULL,
"hq_value" TEXT NOT NULL,
"subsidiary_value" TEXT NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "reference_field_mappings_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "subsidiaries_code_key" ON "subsidiaries"("code");
-- CreateIndex
CREATE UNIQUE INDEX "subsidiaries_routing_value_key" ON "subsidiaries"("routing_value");
-- CreateIndex
CREATE UNIQUE INDEX "replication_rules_business_object_key" ON "replication_rules"("business_object");
-- CreateIndex
CREATE UNIQUE INDEX "mapped_fields_business_object_field_name_key" ON "mapped_fields"("business_object", "field_name");
-- CreateIndex
CREATE UNIQUE INDEX "reference_field_mappings_business_object_field_name_subsidi_key" ON "reference_field_mappings"("business_object", "field_name", "subsidiary_code", "hq_value");
-- AddForeignKey
ALTER TABLE "reference_field_mappings" ADD CONSTRAINT "reference_field_mappings_subsidiary_code_fkey" FOREIGN KEY ("subsidiary_code") REFERENCES "subsidiaries"("code") ON DELETE CASCADE ON UPDATE CASCADE;
@@ -0,0 +1,144 @@
-- CreateEnum
CREATE TYPE "ScenarioDestinationMode" AS ENUM ('fixed', 'payload');
-- DropForeignKey
ALTER TABLE "reference_field_mappings" DROP CONSTRAINT "reference_field_mappings_subsidiary_code_fkey";
-- AlterTable
ALTER TABLE "replication_attempts" DROP COLUMN "subsidiary",
ADD COLUMN "destination_database" TEXT NOT NULL,
ADD COLUMN "origin_database" TEXT NOT NULL,
ADD COLUMN "payload_sent" JSONB,
ADD COLUMN "scenario_id" INTEGER NOT NULL;
-- DropTable
DROP TABLE "mapped_fields";
-- DropTable
DROP TABLE "reference_field_mappings";
-- DropTable
DROP TABLE "replication_rules";
-- DropTable
DROP TABLE "subsidiaries";
-- DropEnum
DROP TYPE "ReplicationRuleType";
-- CreateTable
CREATE TABLE "databases" (
"id" SERIAL NOT NULL,
"code" TEXT NOT NULL,
"name" TEXT NOT NULL,
"base_url" TEXT NOT NULL,
"company_db" TEXT NOT NULL,
"username" TEXT NOT NULL,
"password" TEXT NOT NULL,
"active" BOOLEAN NOT NULL DEFAULT true,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "databases_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "scenarios" (
"id" SERIAL NOT NULL,
"code" TEXT NOT NULL,
"name" TEXT NOT NULL,
"business_object" TEXT NOT NULL,
"origin_database_code" TEXT NOT NULL,
"destination_mode" "ScenarioDestinationMode" NOT NULL,
"key_field" TEXT NOT NULL,
"routing_field" TEXT,
"filter_field" TEXT,
"filter_operator" TEXT,
"filter_value" TEXT,
"active" BOOLEAN NOT NULL DEFAULT true,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "scenarios_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "scenario_destinations" (
"id" SERIAL NOT NULL,
"scenario_id" INTEGER NOT NULL,
"database_code" TEXT NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "scenario_destinations_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "scenario_mapped_fields" (
"id" SERIAL NOT NULL,
"scenario_id" INTEGER NOT NULL,
"field_name" TEXT NOT NULL,
"active" BOOLEAN NOT NULL DEFAULT true,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "scenario_mapped_fields_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "scenario_value_mappings" (
"id" SERIAL NOT NULL,
"scenario_id" INTEGER NOT NULL,
"field_name" TEXT NOT NULL,
"database_code" TEXT NOT NULL,
"source_value" TEXT NOT NULL,
"target_value" TEXT NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "scenario_value_mappings_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "databases_code_key" ON "databases"("code");
-- CreateIndex
CREATE UNIQUE INDEX "scenarios_code_key" ON "scenarios"("code");
-- CreateIndex
CREATE UNIQUE INDEX "scenario_destinations_scenario_id_database_code_key" ON "scenario_destinations"("scenario_id", "database_code");
-- CreateIndex
CREATE UNIQUE INDEX "scenario_mapped_fields_scenario_id_field_name_key" ON "scenario_mapped_fields"("scenario_id", "field_name");
-- CreateIndex
CREATE UNIQUE INDEX "scenario_value_mappings_scenario_id_field_name_database_cod_key" ON "scenario_value_mappings"("scenario_id", "field_name", "database_code", "source_value");
-- CreateIndex
CREATE INDEX "replication_attempts_event_id_idx" ON "replication_attempts"("event_id");
-- CreateIndex
CREATE INDEX "replication_attempts_scenario_id_idx" ON "replication_attempts"("scenario_id");
-- CreateIndex
CREATE INDEX "replication_attempts_attempted_at_idx" ON "replication_attempts"("attempted_at");
-- AddForeignKey
ALTER TABLE "scenarios" ADD CONSTRAINT "scenarios_origin_database_code_fkey" FOREIGN KEY ("origin_database_code") REFERENCES "databases"("code") ON DELETE RESTRICT ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "scenario_destinations" ADD CONSTRAINT "scenario_destinations_scenario_id_fkey" FOREIGN KEY ("scenario_id") REFERENCES "scenarios"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "scenario_destinations" ADD CONSTRAINT "scenario_destinations_database_code_fkey" FOREIGN KEY ("database_code") REFERENCES "databases"("code") ON DELETE RESTRICT ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "scenario_mapped_fields" ADD CONSTRAINT "scenario_mapped_fields_scenario_id_fkey" FOREIGN KEY ("scenario_id") REFERENCES "scenarios"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "scenario_value_mappings" ADD CONSTRAINT "scenario_value_mappings_scenario_id_fkey" FOREIGN KEY ("scenario_id") REFERENCES "scenarios"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "scenario_value_mappings" ADD CONSTRAINT "scenario_value_mappings_database_code_fkey" FOREIGN KEY ("database_code") REFERENCES "databases"("code") ON DELETE RESTRICT ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "replication_attempts" ADD CONSTRAINT "replication_attempts_scenario_id_fkey" FOREIGN KEY ("scenario_id") REFERENCES "scenarios"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
@@ -0,0 +1,15 @@
-- DropForeignKey
ALTER TABLE "scenario_mapped_fields" DROP CONSTRAINT "scenario_mapped_fields_scenario_id_fkey";
-- DropForeignKey
ALTER TABLE "scenario_value_mappings" DROP CONSTRAINT "scenario_value_mappings_database_code_fkey";
-- DropForeignKey
ALTER TABLE "scenario_value_mappings" DROP CONSTRAINT "scenario_value_mappings_scenario_id_fkey";
-- DropTable
DROP TABLE "scenario_mapped_fields";
-- DropTable
DROP TABLE "scenario_value_mappings";
@@ -0,0 +1,7 @@
-- CreateEnum
CREATE TYPE "ScenarioKeyStrategy" AS ENUM ('portable', 'udfLink');
-- AlterTable
ALTER TABLE "scenarios" ADD COLUMN "key_strategy" "ScenarioKeyStrategy" NOT NULL DEFAULT 'portable',
ADD COLUMN "match_field" TEXT;
@@ -0,0 +1,6 @@
-- CreateEnum
CREATE TYPE "ScenarioKeyType" AS ENUM ('string', 'number');
-- AlterTable
ALTER TABLE "scenarios" ADD COLUMN "key_type" "ScenarioKeyType" NOT NULL DEFAULT 'string';
@@ -0,0 +1,23 @@
-- CreateTable
CREATE TABLE "scenario_value_mappings" (
"id" SERIAL NOT NULL,
"scenario_id" INTEGER NOT NULL,
"field_name" TEXT NOT NULL,
"database_code" TEXT NOT NULL,
"source_value" TEXT NOT NULL,
"target_value" TEXT NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "scenario_value_mappings_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "scenario_value_mappings_scenario_id_field_name_database_cod_key" ON "scenario_value_mappings"("scenario_id", "field_name", "database_code", "source_value");
-- AddForeignKey
ALTER TABLE "scenario_value_mappings" ADD CONSTRAINT "scenario_value_mappings_scenario_id_fkey" FOREIGN KEY ("scenario_id") REFERENCES "scenarios"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "scenario_value_mappings" ADD CONSTRAINT "scenario_value_mappings_database_code_fkey" FOREIGN KEY ("database_code") REFERENCES "databases"("code") ON DELETE RESTRICT ON UPDATE CASCADE;
@@ -0,0 +1,5 @@
-- AlterTable
ALTER TABLE "scenarios" DROP COLUMN "filter_field",
DROP COLUMN "filter_operator",
DROP COLUMN "filter_value";
@@ -0,0 +1,3 @@
-- AlterTable
ALTER TABLE "scenarios" ADD COLUMN "destination_business_object" TEXT;
+3
View File
@@ -0,0 +1,3 @@
# Please do not edit this file manually
# It should be added in your version-control system (e.g., Git)
provider = "postgresql"
+177
View File
@@ -0,0 +1,177 @@
// This is your Prisma schema file,
// learn more about it in the docs: https://pris.ly/d/prisma-schema
generator client {
provider = "prisma-client"
output = "../src/generated/prisma"
}
datasource db {
provider = "postgresql"
}
enum EventStatus {
pending
processing
done
partial
failed
filtered // ningún escenario activo para el objeto aceptó el mensaje: no se reintenta
}
enum ScenarioDestinationMode {
fixed // datos maestros: la lista de destinos está fija en ScenarioDestination
payload // transacciones: el destino se resuelve leyendo routingField del payload (no implementado aún, ver docs)
}
// Bandeja de entrada: un registro por evento recibido del webhook de SAP.
// sap-webhook-api inserta acá (además del .json local) apenas recibe la notificación.
model WebhookEvent {
id String @id // el "id" del CloudEvent que manda SAP
businessObject String @map("business_object") // 'BusinessPartners', 'Orders', etc.
operation String // 'Created' | 'Updated'
subject String // CardCode / DocEntry, según el objeto
payload Json
receivedAt DateTime @default(now()) @map("received_at")
status EventStatus @default(pending)
attempts ReplicationAttempt[]
@@map("webhook_events")
}
// Registro único de toda base de datos SAP alcanzable por Service Layer.
// Ya no existe distinción de HQ vs. subsidiaria: cualquier fila puede ser
// origen o destino, según lo que declare cada escenario.
model Database {
id Int @id @default(autoincrement())
code String @unique // mnemónico corto y estable, ej. 'HQ', 'SUB-NORTE'; asignado una sola vez al crear, NO autoincremental
name String
baseUrl String @map("base_url")
companyDb String @map("company_db")
username String
password String
active Boolean @default(true)
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @updatedAt @map("updated_at")
originOfScenarios Scenario[] @relation("ScenarioOrigin")
destinations ScenarioDestination[]
valueMappings ScenarioValueMapping[]
@@map("databases")
}
enum ScenarioKeyStrategy {
portable // el keyField vale lo mismo en origen y destino (ej. CardCode, ItemCode): GET directo por clave
udfLink // SAP asigna la clave por base (ej. SalesEmployeeCode): se busca por un UDF en el destino que guarda la clave de origen
}
enum ScenarioKeyType {
string // la clave va entre comillas en la URL OData: BusinessPartners('C23900')
number // la clave va sin comillas: SalesPersons(5) — mandar comillas en un campo numérico da HTTP 400
}
// Un escenario es una forma completa e independiente de procesar un business
// object: su propio filtro, base origen, base(s) destino y transformación de
// payload. Puede haber varios escenarios activos para el mismo business
// object a la vez (ej. "clientes" y "proveedores" sobre BusinessPartners), y
// un mismo evento puede matchear y ser procesado por más de uno.
//
// El filtro y la transformación del payload NO viven en esta tabla: cada
// escenario tiene un archivo de código en src/scenarios/<code>.ts con
// SIEMPRE las mismas dos partes — filter(record) -> boolean y
// transform(record, ctx) -> record — porque ambos necesitan lógica
// arbitraria (condiciones compuestas, elegir campos) que una tabla no puede
// expresar. Lo que SÍ vive en tabla son los mapeos de valor puntuales que
// transform consulta vía ctx.mapValue(...) dentro de su lógica — ver
// ScenarioValueMapping más abajo y docs/ESCENARIOS.md — pensado para que a
// futuro una UI permita cargarlos sin editar código, aunque el archivo siga
// siendo quien decide cómo usarlos.
model Scenario {
id Int @id @default(autoincrement())
code String @unique // 'BP_CLIENTES', 'BP_PROVEEDORES' — también el nombre del archivo en src/scenarios/
name String
businessObject String @map("business_object") // objeto que dispara el escenario y que se lee del origen (con qué se matchea webhook_events y con qué se hace el GET a la base origen)
destinationBusinessObject String? @map("destination_business_object") // objeto a usar en el GET/POST/PATCH del DESTINO, si es distinto al de origen (ej. origen PurchaseOrders -> destino Orders). null = mismo que businessObject
originDatabaseCode String @map("origin_database_code")
originDatabase Database @relation("ScenarioOrigin", fields: [originDatabaseCode], references: [code])
destinationMode ScenarioDestinationMode @map("destination_mode")
keyField String @map("key_field") // clave del registro para GET/POST/PATCH, ej. 'CardCode', 'SalesEmployeeCode'
keyType ScenarioKeyType @default(string) @map("key_type") // tipo de dato del keyField en el Service Layer (define si va entre comillas en la URL)
keyStrategy ScenarioKeyStrategy @default(portable) @map("key_strategy")
matchField String? @map("match_field") // solo si keyStrategy = udfLink: UDF en el destino que guarda el valor de keyField de origen, ej. 'U_HQ_SlpCode'
routingField String? @map("routing_field") // solo si destinationMode = payload
active Boolean @default(true)
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @updatedAt @map("updated_at")
destinations ScenarioDestination[]
valueMappings ScenarioValueMapping[]
attempts ReplicationAttempt[]
@@map("scenarios")
}
// Destinos fijos de un escenario en modo 'fixed' (datos maestros). En modo
// 'payload' (transacciones) no se usa: el destino se resuelve en runtime.
model ScenarioDestination {
id Int @id @default(autoincrement())
scenarioId Int @map("scenario_id")
scenario Scenario @relation(fields: [scenarioId], references: [id], onDelete: Cascade)
databaseCode String @map("database_code")
database Database @relation(fields: [databaseCode], references: [code])
createdAt DateTime @default(now()) @map("created_at")
@@unique([scenarioId, databaseCode])
@@map("scenario_destinations")
}
// Tabla de equivalencias que el archivo de transformación del escenario
// consulta vía ctx.mapValue(fieldName, sourceValue) — el código decide QUÉ
// campo mapear y QUÉ hacer si falta (fallar, dejar sin traducir, etc.); acá
// solo vive el dato. Escopada por escenario, no global por objeto+campo,
// porque el mismo campo puede tener dominios de valores distintos según el
// escenario (ej. GroupCode de clientes vs. de proveedores). Pensada para
// que a futuro una UI la administre sin tocar código.
model ScenarioValueMapping {
id Int @id @default(autoincrement())
scenarioId Int @map("scenario_id")
scenario Scenario @relation(fields: [scenarioId], references: [id], onDelete: Cascade)
fieldName String @map("field_name")
databaseCode String @map("database_code")
database Database @relation(fields: [databaseCode], references: [code])
sourceValue String @map("source_value")
targetValue String @map("target_value")
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @updatedAt @map("updated_at")
@@unique([scenarioId, fieldName, databaseCode, sourceValue])
@@map("scenario_value_mappings")
}
// Un registro por cada intento de réplica de un evento, dentro de UN
// escenario puntual, hacia UNA base destino. origin_database y
// destination_database son texto libre (sin FK) a propósito: es un log de
// auditoría inmutable que no debe verse afectado si la base o el escenario
// se borran o renombran después.
model ReplicationAttempt {
id Int @id @default(autoincrement())
event WebhookEvent @relation(fields: [eventId], references: [id])
eventId String @map("event_id")
scenario Scenario @relation(fields: [scenarioId], references: [id])
scenarioId Int @map("scenario_id")
originDatabase String @map("origin_database")
destinationDatabase String @map("destination_database") // 'N/A' si falló antes de resolver destino (ej. error leyendo el origen)
verbUsed String @map("verb_used") // 'POST' | 'PATCH' | 'SKIPPED' | 'N/A'
httpStatus Int? @map("http_status")
errorMessage String? @map("error_message")
payloadSent Json? @map("payload_sent") // payload ya transformado enviado al destino; se guarda en éxito Y en falla
resultKey String? @map("result_key") // clave propia del destino cuando difiere de la de origen (keyStrategy=udfLink) o DocEntry (transacciones); null si no aplica
attemptedAt DateTime @default(now()) @map("attempted_at")
@@index([eventId])
@@index([scenarioId])
@@index([attemptedAt])
@@map("replication_attempts")
}
+315
View File
@@ -0,0 +1,315 @@
// CLI de administración de la configuración de replicación (bases de datos,
// escenarios, destinos y mapeos de valor). Reemplaza la edición manual de
// SUBSIDIARIES/HQ_* en .env: los cambios acá tienen efecto en el próximo
// tick del worker, sin reiniciar el proceso.
//
// El filtro y la transformación del payload de un escenario son un archivo
// de código en src/scenarios/<code>.ts (ver docs/ESCENARIOS.md) — SIEMPRE
// con las mismas dos partes (filter + transform). `scenario:add` genera uno
// de partida la primera vez que se crea un escenario nuevo. Los mapeos de
// valor (scenario:mapping:*) SÍ se administran acá: son datos que ese
// archivo consulta vía ctx.mapValue(...) dentro de transform, pero la
// decisión de qué campo mapear y qué hacer si falta el mapeo queda siempre
// en el código.
//
// Uso:
// tsx scripts/manage-config.ts database:add <code> <name> <baseUrl> <companyDb> <user> <pass>
// tsx scripts/manage-config.ts database:disable <code>
// tsx scripts/manage-config.ts database:list
//
// tsx scripts/manage-config.ts scenario:add <code> <name> <businessObject> <originDbCode> <keyField>
// [--key-strategy udfLink --match-field U_HQ_SlpCode] [--key-type number]
// [--destination-mode payload --routing-field U_ITC_Target]
// tsx scripts/manage-config.ts scenario:rename <oldCode> <newCode>
// tsx scripts/manage-config.ts scenario:disable <code>
// tsx scripts/manage-config.ts scenario:list
//
// tsx scripts/manage-config.ts scenario:add-destination <scenarioCode> <databaseCode>
// tsx scripts/manage-config.ts scenario:remove-destination <scenarioCode> <databaseCode>
//
// tsx scripts/manage-config.ts scenario:mapping:add <scenarioCode> <fieldName> <databaseCode> <sourceValue> <targetValue>
// tsx scripts/manage-config.ts scenario:mapping:list <scenarioCode> <fieldName>
import 'dotenv/config';
import { existsSync, mkdirSync, renameSync, writeFileSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { PrismaPg } from '@prisma/adapter-pg';
import { PrismaClient } from '../src/generated/prisma/client.ts';
import { scenarioFileSlug } from '../src/scenarioTransform.ts';
const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL });
const prisma = new PrismaClient({ adapter });
const scenariosDir = fileURLToPath(new URL('../src/scenarios/', import.meta.url));
function scenarioFilePath(code: string): string {
return `${scenariosDir}${scenarioFileSlug(code)}.ts`;
}
function scaffoldTransformFile(code: string): string | null {
const path = scenarioFilePath(code);
if (existsSync(path)) return null;
mkdirSync(scenariosDir, { recursive: true });
writeFileSync(
path,
`// Escenario ${code} — completar según corresponda (ver docs/ESCENARIOS.md).
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: decide si este escenario aplica al registro ---
// Puede ser async: ctx.get(path) permite consultar la base ORIGEN (ej. si
// el filtro depende de un dato que no viene en el payload del evento).
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean | Promise<boolean> {
return true; // reemplazar con la condición real, o dejar así para "sin filtro"
}
// --- Transformación: arma el payload para un destino puntual ---
// Se llama una vez por destino: ctx.destinationCode permite transformar
// distinto según a cuál le toque. Puede lanzar una excepción para marcar
// el intento como SKIPPED (ej. un valor sin mapeo cargado).
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
// --- Mapeo: ctx.mapValue(fieldName, sourceValue) consulta scenario_value_mappings
// (cargados con 'scenario:mapping:add') para ESTE escenario y destino ---
return { ...record };
}
`,
);
return path;
}
function fail(message: string): never {
console.error(`Error: ${message}`);
process.exit(1);
}
function flag(args: string[], name: string): string | undefined {
const index = args.indexOf(`--${name}`);
return index === -1 ? undefined : args[index + 1];
}
async function requireScenario(code: string) {
const scenario = await prisma.scenario.findUnique({ where: { code } });
if (!scenario) fail(`no existe el escenario '${code}'`);
return scenario;
}
async function main() {
const [command, ...args] = process.argv.slice(2);
switch (command) {
case 'database:add': {
const [code, name, baseUrl, companyDb, username, password] = args;
if (!code || !name || !baseUrl || !companyDb || !username || !password) {
fail('database:add <code> <name> <baseUrl> <companyDb> <user> <pass>');
}
const db = await prisma.database.upsert({
where: { code },
create: { code, name, baseUrl, companyDb, username, password },
update: { name, baseUrl, companyDb, username, password, active: true },
});
console.log(`Base de datos ${db.code} guardada (id=${db.id}).`);
break;
}
case 'database:disable': {
const [code] = args;
if (!code) fail('database:disable <code>');
await prisma.database.update({ where: { code }, data: { active: false } });
console.log(`Base de datos ${code} desactivada.`);
break;
}
case 'database:list': {
const dbs = await prisma.database.findMany({ orderBy: { code: 'asc' } });
console.table(dbs.map((d) => ({ code: d.code, name: d.name, baseUrl: d.baseUrl, companyDb: d.companyDb, active: d.active })));
break;
}
case 'scenario:add': {
const [code, name, businessObject, originDatabaseCode, keyField] = args;
if (!code || !name || !businessObject || !originDatabaseCode || !keyField) {
fail(
'scenario:add <code> <name> <businessObject> <originDbCode> <keyField> ' +
'[--key-strategy udfLink --match-field U_HQ_SlpCode] [--key-type number] ' +
'[--destination-mode payload --routing-field U_ITC_Target] ' +
'[--destination-object Orders]',
);
}
const keyStrategyFlag = flag(args, 'key-strategy') ?? 'portable';
if (keyStrategyFlag !== 'portable' && keyStrategyFlag !== 'udfLink') {
fail("--key-strategy debe ser 'portable' o 'udfLink'");
}
const matchField = flag(args, 'match-field') ?? null;
if (keyStrategyFlag === 'udfLink' && !matchField) {
fail('--key-strategy udfLink necesita --match-field <UDF>');
}
const keyTypeFlag = flag(args, 'key-type') ?? 'string';
if (keyTypeFlag !== 'string' && keyTypeFlag !== 'number') {
fail("--key-type debe ser 'string' o 'number' (numérico va sin comillas en la URL OData, ej. SalesEmployeeCode)");
}
const destinationModeFlag = flag(args, 'destination-mode') ?? 'fixed';
if (destinationModeFlag !== 'fixed' && destinationModeFlag !== 'payload') {
fail("--destination-mode debe ser 'fixed' o 'payload'");
}
const routingField = flag(args, 'routing-field') ?? null;
if (destinationModeFlag === 'payload' && !routingField) {
fail('--destination-mode payload necesita --routing-field <UDF> (campo en el origen con el code de la base destino)');
}
const destinationObject = flag(args, 'destination-object') ?? null;
const existed = (await prisma.scenario.findUnique({ where: { code } })) !== null;
const scenario = await prisma.scenario.upsert({
where: { code },
create: {
code,
name,
businessObject,
destinationBusinessObject: destinationObject,
originDatabaseCode,
destinationMode: destinationModeFlag,
routingField,
keyField,
keyType: keyTypeFlag,
keyStrategy: keyStrategyFlag,
matchField,
},
update: {
name,
businessObject,
destinationBusinessObject: destinationObject,
originDatabaseCode,
destinationMode: destinationModeFlag,
routingField,
keyField,
keyType: keyTypeFlag,
keyStrategy: keyStrategyFlag,
matchField,
active: true,
},
});
console.log(`Escenario ${scenario.code} guardado (id=${scenario.id}).`);
if (!existed) {
const scaffolded = scaffoldTransformFile(scenario.code);
if (scaffolded) console.log(`Archivo de transformación creado: ${scaffolded} (editalo antes de usar el escenario).`);
}
break;
}
case 'scenario:rename': {
const [oldCode, newCode] = args;
if (!oldCode || !newCode) fail('scenario:rename <oldCode> <newCode>');
const scenario = await requireScenario(oldCode);
const oldPath = scenarioFilePath(oldCode);
const newPath = scenarioFilePath(newCode);
if (existsSync(newPath)) fail(`ya existe un archivo en ${newPath}`);
await prisma.scenario.update({ where: { id: scenario.id }, data: { code: newCode } });
if (existsSync(oldPath)) {
renameSync(oldPath, newPath);
console.log(`Archivo renombrado: ${oldPath} -> ${newPath}`);
} else {
console.warn(`Aviso: no se encontró ${oldPath} — creá ${newPath} a mano antes de usar el escenario.`);
}
console.log(`Escenario ${oldCode} renombrado a ${newCode}.`);
break;
}
case 'scenario:disable': {
const [code] = args;
if (!code) fail('scenario:disable <code>');
await prisma.scenario.update({ where: { code }, data: { active: false } });
console.log(`Escenario ${code} desactivado.`);
break;
}
case 'scenario:list': {
const scenarios = await prisma.scenario.findMany({
orderBy: { code: 'asc' },
include: { destinations: true },
});
console.table(
scenarios.map((s) => ({
code: s.code,
businessObject: s.destinationBusinessObject ? `${s.businessObject} -> ${s.destinationBusinessObject}` : s.businessObject,
origin: s.originDatabaseCode,
destinations:
s.destinationMode === 'payload' ? `(via payload: ${s.routingField})` : s.destinations.map((d) => d.databaseCode).join(', '),
keyField: s.keyStrategy === 'udfLink' ? `${s.keyField} (via ${s.matchField})` : s.keyField,
archivo: `src/scenarios/${scenarioFileSlug(s.code)}.ts`,
active: s.active,
})),
);
break;
}
case 'scenario:add-destination': {
const [scenarioCode, databaseCode] = args;
if (!scenarioCode || !databaseCode) fail('scenario:add-destination <scenarioCode> <databaseCode>');
const scenario = await requireScenario(scenarioCode);
await prisma.scenarioDestination.upsert({
where: { scenarioId_databaseCode: { scenarioId: scenario.id, databaseCode } },
create: { scenarioId: scenario.id, databaseCode },
update: {},
});
console.log(`Destino ${databaseCode} agregado a escenario ${scenarioCode}.`);
break;
}
case 'scenario:remove-destination': {
const [scenarioCode, databaseCode] = args;
if (!scenarioCode || !databaseCode) fail('scenario:remove-destination <scenarioCode> <databaseCode>');
const scenario = await requireScenario(scenarioCode);
await prisma.scenarioDestination.delete({
where: { scenarioId_databaseCode: { scenarioId: scenario.id, databaseCode } },
});
console.log(`Destino ${databaseCode} quitado de escenario ${scenarioCode}.`);
break;
}
case 'scenario:mapping:add': {
const [scenarioCode, fieldName, databaseCode, sourceValue, targetValue] = args;
if (!scenarioCode || !fieldName || !databaseCode || !sourceValue || !targetValue) {
fail('scenario:mapping:add <scenarioCode> <fieldName> <databaseCode> <sourceValue> <targetValue>');
}
const scenario = await requireScenario(scenarioCode);
await prisma.scenarioValueMapping.upsert({
where: {
scenarioId_fieldName_databaseCode_sourceValue: { scenarioId: scenario.id, fieldName, databaseCode, sourceValue },
},
create: { scenarioId: scenario.id, fieldName, databaseCode, sourceValue, targetValue },
update: { targetValue },
});
console.log(`Mapeo ${scenarioCode}/${fieldName} [${databaseCode}]: ${sourceValue} -> ${targetValue} guardado.`);
break;
}
case 'scenario:mapping:list': {
const [scenarioCode, fieldName] = args;
if (!scenarioCode || !fieldName) fail('scenario:mapping:list <scenarioCode> <fieldName>');
const scenario = await requireScenario(scenarioCode);
const mappings = await prisma.scenarioValueMapping.findMany({
where: { scenarioId: scenario.id, fieldName },
orderBy: [{ databaseCode: 'asc' }, { sourceValue: 'asc' }],
});
console.table(mappings.map((m) => ({ database: m.databaseCode, sourceValue: m.sourceValue, targetValue: m.targetValue })));
break;
}
default:
fail(
`comando desconocido: ${command ?? '(ninguno)'}\nComandos disponibles: database:add, database:disable, database:list, ` +
'scenario:add, scenario:rename, scenario:disable, scenario:list, scenario:add-destination, scenario:remove-destination, ' +
'scenario:mapping:add, scenario:mapping:list',
);
}
}
main()
.catch((err) => {
console.error(err);
process.exitCode = 1;
})
.finally(() => prisma.$disconnect());
+16
View File
@@ -0,0 +1,16 @@
import 'dotenv/config';
export interface ServiceLayerTarget {
name: string;
baseUrl: string;
companyDb: string;
username: string;
password: string;
}
// Ya no hay una base "HQ" hardcodeada en .env: toda base (incluida la casa
// matriz) es una fila más en la tabla `databases`, y cada escenario declara
// cuál usa como origen. Ver src/replicationConfig.ts y scripts/manage-config.ts.
export const config = {
pollIntervalSeconds: Number(process.env.POLL_INTERVAL_SECONDS ?? 30),
};
+33
View File
@@ -0,0 +1,33 @@
import cron from 'node-cron';
import { PrismaPg } from '@prisma/adapter-pg';
import { PrismaClient } from './generated/prisma/client.ts';
import { config } from './config.ts';
import { loadReplicationConfig } from './replicationConfig.ts';
import { processPendingEvents } from './replicateEvents.ts';
const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL });
const prisma = new PrismaClient({ adapter });
async function runOnce(): Promise<void> {
try {
const cfg = await loadReplicationConfig(prisma);
await processPendingEvents(prisma, cfg);
} catch (err) {
console.error('[worker] Error procesando eventos pendientes:', err);
}
}
const runOnceOnly = process.argv.includes('--once');
if (runOnceOnly) {
await runOnce();
await prisma.$disconnect();
process.exit(0);
} else {
const cronExpression = `*/${config.pollIntervalSeconds} * * * * *`;
console.log(
`[worker] sap-intercompany-worker arrancó. Revisando eventos pendientes cada ${config.pollIntervalSeconds}s.`,
);
cron.schedule(cronExpression, runOnce);
await runOnce(); // primera corrida inmediata, sin esperar el primer tick
}
+418
View File
@@ -0,0 +1,418 @@
import type { PrismaClient } from './generated/prisma/client.ts';
import { Prisma } from './generated/prisma/client.ts';
import { ServiceLayerClient } from './serviceLayer.ts';
import type { ServiceLayerTarget } from './config.ts';
import {
applyScenarioTransform,
buildScenarioFilterContext,
buildScenarioTransformContext,
loadScenarioModule,
} from './scenarioTransform.ts';
import type { ReplicationConfig } from './replicationConfig.ts';
// Campos que no se deben reenviar tal cual al hacer PATCH/POST en el
// destino (metadata propia de la lectura, no del dato en sí).
const FIELDS_TO_STRIP = ['@odata.context', '@odata.etag'];
type Database = ReplicationConfig['databases'][number];
type ScenarioRow = ReplicationConfig['scenarios'][number];
function cleanRecord(record: Record<string, unknown>): Record<string, unknown> {
const clean = { ...record };
for (const field of FIELDS_TO_STRIP) delete clean[field];
return clean;
}
function toServiceLayerTarget(db: Database): ServiceLayerTarget {
return { name: db.code, baseUrl: db.baseUrl, companyDb: db.companyDb, username: db.username, password: db.password };
}
function escapeODataString(value: string): string {
return value.replace(/'/g, "''");
}
// El CloudEvent trae "source": "/default/sap.b1/<CompanyDB>" — de qué base
// SAP salió la notificación. Un escenario solo debe reaccionar a eventos que
// realmente vinieron de SU base origen configurada: si una base DESTINO
// también tiene su propia suscripción de webhook activa (para otro fin), un
// PATCH nuestro ahí dispara una notificación que, sin este chequeo, el
// motor volvería a tomar como si fuera un cambio nuevo en el origen —
// reprocesándolo, volviendo a hacer PATCH, y así indefinidamente.
function extractSourceCompanyDb(payload: unknown): string | undefined {
if (!payload || typeof payload !== 'object') return undefined;
const source = (payload as Record<string, unknown>).source;
if (typeof source !== 'string') return undefined;
const segments = source.split('/').filter(Boolean);
return segments.length > 0 ? segments[segments.length - 1] : undefined;
}
// Un campo numérico va sin comillas en la URL OData (SalesPersons(5)); uno de
// texto va entre comillas (BusinessPartners('C23900')). Mandar comillas en un
// campo numérico devuelve HTTP 400.
function formatKeySegment(key: string, keyType: ScenarioRow['keyType']): string {
return keyType === 'number' ? key : `'${escapeODataString(key)}'`;
}
type ExistenceResult = { found: true; destinationKey: string } | { found: false } | { error: string };
// Resuelve si el registro ya existe en el destino, y con qué clave.
// - keyStrategy='portable': la clave vale lo mismo en origen y destino (CardCode, ItemCode) -> GET directo.
// - keyStrategy='udfLink': SAP asigna la clave por base (ej. SlpCode) -> se busca por el UDF (matchField)
// que guarda el valor de la clave de origen, y la clave real del destino sale del registro encontrado.
async function resolveDestinationRecord(
client: ServiceLayerClient,
destinationBusinessObject: string,
scenario: ScenarioRow,
originKey: string,
): Promise<ExistenceResult> {
if (scenario.keyStrategy === 'portable') {
const res = await client.get(`/${destinationBusinessObject}(${formatKeySegment(originKey, scenario.keyType)})`);
if (res.status === 200) return { found: true, destinationKey: originKey };
if (res.status === 404) return { found: false };
return { error: `Respuesta inesperada al chequear existencia: HTTP ${res.status}` };
}
if (!scenario.matchField) {
return { error: `Escenario '${scenario.code}' es keyStrategy=udfLink pero no tiene matchField configurado` };
}
const filter = `${scenario.matchField} eq '${escapeODataString(originKey)}'`;
const res = await client.get(`/${destinationBusinessObject}?$filter=${encodeURIComponent(filter)}`);
if (res.status !== 200) {
return { error: `Respuesta inesperada al buscar por ${scenario.matchField}: HTTP ${res.status}` };
}
const matches: Array<Record<string, unknown>> = Array.isArray(res.data?.value) ? res.data.value : [];
if (matches.length === 0) return { found: false };
const destinationKey = matches[0][scenario.keyField];
if (destinationKey === undefined || destinationKey === null) {
return { error: `Se encontró un registro por ${scenario.matchField} pero no tiene el campo clave '${scenario.keyField}'` };
}
return { found: true, destinationKey: String(destinationKey) };
}
// Arma el payload para UN destino puntual, resuelve si existe (PATCH) o no
// (POST), y registra el intento. Común a destinationMode='fixed' (destino
// de una lista fija) y 'payload' (destino resuelto desde el propio mensaje)
// — a esta altura ya no importa de dónde salió destCode.
async function replicateToDestination(
prisma: PrismaClient,
event: { id: string },
scenario: ScenarioRow,
destinationBusinessObject: string,
key: string,
originDb: Database,
sourceRecord: Record<string, unknown>,
scenarioModule: Awaited<ReturnType<typeof loadScenarioModule>>,
cfg: ReplicationConfig,
destCode: string,
): Promise<boolean> {
const destDb = cfg.databases.find((d) => d.code === destCode);
if (!destDb) {
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: destCode,
verbUsed: 'N/A',
httpStatus: null,
errorMessage: `Base destino '${destCode}' no existe o está inactiva`,
},
});
return false;
}
let transformedRecord: Record<string, unknown>;
try {
const ctx = buildScenarioTransformContext(destDb.code, scenario.id, cfg.valueMappings);
transformedRecord = applyScenarioTransform(scenarioModule.transform, sourceRecord, ctx);
} catch (err) {
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: destDb.code,
verbUsed: 'SKIPPED',
httpStatus: null,
errorMessage: err instanceof Error ? err.message : String(err),
},
});
return false;
}
const client = new ServiceLayerClient(toServiceLayerTarget(destDb));
// En keyStrategy=udfLink, el UDF con la clave de origen viaja SIEMPRE en el
// payload (tanto al crear como al actualizar) para que la relación quede
// registrada en SAP mismo, no solo en nuestra config.
const payloadToSend =
scenario.keyStrategy === 'udfLink' && scenario.matchField
? { ...transformedRecord, [scenario.matchField]: key }
: transformedRecord;
let verbUsed: 'POST' | 'PATCH' = 'POST';
let httpStatus: number | null = null;
let errorMessage: string | null = null;
let resultKey: string | null = null;
try {
const existence = await resolveDestinationRecord(client, destinationBusinessObject, scenario, key);
if ('error' in existence) {
errorMessage = existence.error;
} else if (existence.found) {
verbUsed = 'PATCH';
resultKey = scenario.keyStrategy === 'udfLink' ? existence.destinationKey : null;
const res = await client.patch(
`/${destinationBusinessObject}(${formatKeySegment(existence.destinationKey, scenario.keyType)})`,
payloadToSend,
);
httpStatus = res.status;
if (res.status >= 400) errorMessage = JSON.stringify(res.data);
} else {
verbUsed = 'POST';
const res = await client.post(`/${destinationBusinessObject}`, payloadToSend);
httpStatus = res.status;
if (res.status >= 400) {
errorMessage = JSON.stringify(res.data);
} else if (scenario.keyStrategy === 'udfLink') {
const createdKey = (res.data as Record<string, unknown> | undefined)?.[scenario.keyField];
resultKey = createdKey !== undefined && createdKey !== null ? String(createdKey) : null;
}
}
} catch (err) {
errorMessage = err instanceof Error ? err.message : String(err);
}
const success = httpStatus !== null && httpStatus >= 200 && httpStatus < 300;
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: destDb.code,
verbUsed,
httpStatus,
errorMessage,
payloadSent: payloadToSend as Prisma.InputJsonValue,
resultKey,
},
});
return success;
}
export async function processPendingEvents(prisma: PrismaClient, cfg: ReplicationConfig): Promise<void> {
const businessObjects = [...new Set(cfg.scenarios.map((s) => s.businessObject))];
for (const businessObject of businessObjects) {
const events = await prisma.webhookEvent.findMany({
where: { businessObject, status: 'pending' },
orderBy: { receivedAt: 'asc' },
});
if (events.length === 0) continue;
const scenarios = cfg.scenarios.filter((s) => s.businessObject === businessObject);
console.log(`[replicate] ${events.length} evento(s) de ${businessObject} pendientes, ${scenarios.length} escenario(s) activo(s)`);
for (const event of events) {
await prisma.webhookEvent.update({ where: { id: event.id }, data: { status: 'processing' } });
const key = event.subject;
const sourceCompanyDb = extractSourceCompanyDb(event.payload);
let matchedAnyScenario = false;
let totalAttempts = 0;
let successAttempts = 0;
for (const scenario of scenarios) {
const originDb = cfg.databases.find((d) => d.code === scenario.originDatabaseCode);
if (originDb && sourceCompanyDb && sourceCompanyDb !== originDb.companyDb) {
console.log(
`[${event.id}] escenario '${scenario.code}' no aplica: el evento vino de '${sourceCompanyDb}', no de la base origen configurada '${originDb.companyDb}' (${originDb.code})`,
);
continue;
}
if (!originDb) {
console.error(`[${event.id}] escenario '${scenario.code}': base origen '${scenario.originDatabaseCode}' no existe o está inactiva`);
matchedAnyScenario = true;
totalAttempts++;
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: scenario.originDatabaseCode,
destinationDatabase: 'N/A',
verbUsed: 'N/A',
httpStatus: null,
errorMessage: `Base origen '${scenario.originDatabaseCode}' no existe o está inactiva`,
},
});
continue;
}
const origin = new ServiceLayerClient(toServiceLayerTarget(originDb));
let sourceRecord: Record<string, unknown>;
try {
const res = await origin.get(`/${businessObject}(${formatKeySegment(key, scenario.keyType)})`);
if (res.status !== 200) {
throw new Error(`HTTP ${res.status} leyendo ${key} de ${originDb.code}`);
}
sourceRecord = cleanRecord(res.data);
} catch (err) {
console.error(`[${event.id}] escenario '${scenario.code}': error leyendo ${key} de ${originDb.code}:`, err);
matchedAnyScenario = true;
totalAttempts++;
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: 'N/A',
verbUsed: 'N/A',
httpStatus: null,
errorMessage: err instanceof Error ? err.message : String(err),
},
});
continue;
}
let scenarioModule: Awaited<ReturnType<typeof loadScenarioModule>>;
try {
scenarioModule = await loadScenarioModule(scenario.code);
} catch (err) {
console.error(`[${event.id}] escenario '${scenario.code}': ${err instanceof Error ? err.message : String(err)}`);
matchedAnyScenario = true;
totalAttempts++;
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: 'N/A',
verbUsed: 'N/A',
httpStatus: null,
errorMessage: err instanceof Error ? err.message : String(err),
},
});
continue;
}
let filterPasses: boolean;
try {
const filterCtx = buildScenarioFilterContext(originDb.code, (path) => origin.get(path));
filterPasses = await scenarioModule.filter(sourceRecord, filterCtx);
} catch (err) {
console.error(`[${event.id}] escenario '${scenario.code}': error en filter():`, err);
matchedAnyScenario = true;
totalAttempts++;
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: 'N/A',
verbUsed: 'N/A',
httpStatus: null,
errorMessage: `Error en filter(): ${err instanceof Error ? err.message : String(err)}`,
},
});
continue;
}
if (!filterPasses) {
console.log(`[${event.id}] escenario '${scenario.code}' no aplica: filter() devolvió false`);
continue;
}
matchedAnyScenario = true;
// Por default el destino usa el mismo objeto que el origen — pero un
// escenario puede pedir uno distinto (ej. origen PurchaseOrders ->
// destino Orders, típico de comercio intercompañía).
const destinationBusinessObject = scenario.destinationBusinessObject ?? businessObject;
if (scenario.destinationMode === 'fixed') {
const destinations = cfg.destinations.filter((d) => d.scenarioId === scenario.id);
for (const dest of destinations) {
totalAttempts++;
const ok = await replicateToDestination(
prisma,
event,
scenario,
destinationBusinessObject,
key,
originDb,
sourceRecord,
scenarioModule,
cfg,
dest.databaseCode,
);
if (ok) successAttempts++;
}
} else {
// destinationMode='payload': el destino no está en config, se
// resuelve leyendo routingField del registro recién leído del origen.
if (!scenario.routingField) {
totalAttempts++;
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: 'N/A',
verbUsed: 'N/A',
httpStatus: null,
errorMessage: `Escenario '${scenario.code}' es destinationMode=payload pero no tiene routingField configurado`,
},
});
} else {
const rawTarget = sourceRecord[scenario.routingField];
const targetCode = typeof rawTarget === 'string' ? rawTarget.trim() : rawTarget != null ? String(rawTarget) : '';
totalAttempts++;
if (!targetCode) {
await prisma.replicationAttempt.create({
data: {
eventId: event.id,
scenarioId: scenario.id,
originDatabase: originDb.code,
destinationDatabase: 'N/A',
verbUsed: 'N/A',
httpStatus: null,
errorMessage: `El campo de ruteo '${scenario.routingField}' viene vacío en el registro de origen`,
},
});
} else {
const ok = await replicateToDestination(
prisma,
event,
scenario,
destinationBusinessObject,
key,
originDb,
sourceRecord,
scenarioModule,
cfg,
targetCode,
);
if (ok) successAttempts++;
}
}
}
}
const finalStatus = !matchedAnyScenario
? 'filtered'
: totalAttempts === 0
? 'failed'
: successAttempts === totalAttempts
? 'done'
: successAttempts === 0
? 'failed'
: 'partial';
await prisma.webhookEvent.update({ where: { id: event.id }, data: { status: finalStatus } });
console.log(`[${event.id}] ${key}: ${successAttempts}/${totalAttempts} intentos OK -> ${finalStatus}`);
}
}
}
+24
View File
@@ -0,0 +1,24 @@
import type { PrismaClient } from './generated/prisma/client.ts';
// Config leída de Postgres al principio de cada tick del cron (no hace falta
// TTL cache aparte: el propio POLL_INTERVAL_SECONDS ya es el punto de refresco).
// La transformación de payload NO está acá: vive en código, en
// src/scenarios/<code>.ts (ver src/scenarioTransform.ts). valueMappings sí
// es dato — ese código la consulta vía ctx.mapValue(...).
export interface ReplicationConfig {
databases: Awaited<ReturnType<PrismaClient['database']['findMany']>>;
scenarios: Awaited<ReturnType<PrismaClient['scenario']['findMany']>>;
destinations: Awaited<ReturnType<PrismaClient['scenarioDestination']['findMany']>>;
valueMappings: Awaited<ReturnType<PrismaClient['scenarioValueMapping']['findMany']>>;
}
export async function loadReplicationConfig(prisma: PrismaClient): Promise<ReplicationConfig> {
const [databases, scenarios, destinations, valueMappings] = await Promise.all([
prisma.database.findMany({ where: { active: true } }),
prisma.scenario.findMany({ where: { active: true } }),
prisma.scenarioDestination.findMany(),
prisma.scenarioValueMapping.findMany(),
]);
return { databases, scenarios, destinations, valueMappings };
}
+114
View File
@@ -0,0 +1,114 @@
// Cada escenario tiene un archivo en src/scenarios/ con SIEMPRE las mismas
// dos partes. El nombre del archivo se deriva del `code` de la fila en
// `scenarios` pasándolo a minúsculas (scenarioFileSlug) — el code en sí
// queda en mayúsculas en la DB/CLI/logs (es el identificador legible), pero
// como nombre de archivo TS en mayúsculas no es una convención usual:
//
// export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean | Promise<boolean>
// export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown>
//
// `filter` decide si el escenario aplica a ESE registro puntual (true =
// aplica). Reemplaza lo que antes era un filtro declarativo en la tabla
// scenarios (un solo campo/operador/valor) — como código, admite cualquier
// condición (AND/OR, varios campos, lo que haga falta), incluida una
// llamada extra al Service Layer vía ctx.get(...) cuando el dato para
// filtrar no viene en el payload del evento (típico en transacciones: una
// Factura trae el CardCode del cliente, pero el filtro depende de un campo
// del BusinessPartner, no de la Factura misma).
//
// `transform` arma el payload a enviar a un destino puntual — se llama una
// vez por destino (ctx.destinationCode indica cuál) y usa
// ctx.mapValue(fieldName, sourceValue) para resolver los mapeos de valor
// cargados en Postgres (ver docs/ESCENARIOS.md) — esa es la parte de
// "mapeo" dentro de la transformación: el archivo decide QUÉ campo mapear
// y QUÉ hacer si falta el valor, el dato en sí vive en `scenario_value_mappings`.
export interface ScenarioFilterContext {
originCode: string;
// GET arbitrario contra la base ORIGEN del escenario, reusando la misma
// sesión ya autenticada que trajo el registro principal (no abre un login
// nuevo). Sirve tanto para un GET por clave ('/BusinessPartners(...)')
// como para una consulta con $filter.
get(path: string): Promise<{ status: number; data: any }>;
}
export interface ScenarioTransformContext {
destinationCode: string;
// Busca en scenario_value_mappings (para ESTE escenario y ESTE destino) el
// valor cargado para fieldName=sourceValue. undefined si no hay mapeo —
// el archivo decide qué hacer (tirar error, dejar sin traducir, etc.), acá
// no se asume nada.
mapValue(fieldName: string, sourceValue: string): string | undefined;
}
export type ScenarioFilterFn = (record: Record<string, unknown>, ctx: ScenarioFilterContext) => boolean | Promise<boolean>;
export type ScenarioTransformFn = (record: Record<string, unknown>, ctx: ScenarioTransformContext) => Record<string, unknown>;
export interface ScenarioModule {
filter: ScenarioFilterFn;
transform: ScenarioTransformFn;
}
// Nombre de archivo a partir del code de un escenario (ej. 'BP_PROVEEDORES'
// -> 'bp_proveedores'). Usado tanto para cargar el módulo en runtime como
// para generar el scaffold en manage-config.ts — ambos deben usar esta
// misma función para no desincronizarse.
export function scenarioFileSlug(scenarioCode: string): string {
return scenarioCode.toLowerCase();
}
export async function loadScenarioModule(scenarioCode: string): Promise<ScenarioModule> {
const fileSlug = scenarioFileSlug(scenarioCode);
const moduleUrl = new URL(`./scenarios/${fileSlug}.ts`, import.meta.url);
let mod: { filter?: unknown; transform?: unknown };
try {
mod = await import(moduleUrl.href);
} catch (err) {
throw new Error(
`No existe (o no se pudo cargar) el archivo de escenario 'src/scenarios/${fileSlug}.ts': ${
err instanceof Error ? err.message : String(err)
}`,
);
}
if (typeof mod.filter !== 'function') {
throw new Error(`'src/scenarios/${fileSlug}.ts' no exporta una función 'filter'`);
}
if (typeof mod.transform !== 'function') {
throw new Error(`'src/scenarios/${fileSlug}.ts' no exporta una función 'transform'`);
}
return { filter: mod.filter as ScenarioFilterFn, transform: mod.transform as ScenarioTransformFn };
}
export function buildScenarioFilterContext(
originCode: string,
get: (path: string) => Promise<{ status: number; data: any }>,
): ScenarioFilterContext {
return { originCode, get };
}
export function buildScenarioTransformContext(
destinationCode: string,
scenarioId: number,
valueMappings: Array<{ scenarioId: number; databaseCode: string; fieldName: string; sourceValue: string; targetValue: string }>,
): ScenarioTransformContext {
return {
destinationCode,
mapValue(fieldName, sourceValue) {
const mapping = valueMappings.find(
(m) => m.scenarioId === scenarioId && m.databaseCode === destinationCode && m.fieldName === fieldName && m.sourceValue === sourceValue,
);
return mapping?.targetValue;
},
};
}
export function applyScenarioTransform(
transform: ScenarioTransformFn,
record: Record<string, unknown>,
ctx: ScenarioTransformContext,
): Record<string, unknown> {
const result = transform({ ...record }, ctx);
if (typeof result !== 'object' || result === null || Array.isArray(result)) {
throw new Error(`El transform de este escenario devolvió algo que no es un objeto JSON para destino '${ctx.destinationCode}'`);
}
return result;
}
+37
View File
@@ -0,0 +1,37 @@
// Escenario BP_PROVEEDORES: BusinessPartners con CardType=cSupplier,
// origen HQ01, destinos SUB_01/SUB_02 — ver docs/ESCENARIOS.md.
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: decide si este escenario aplica al registro ---
// No necesita ctx acá: CardType ya viene en el payload de BusinessPartners.
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean {
return record.CardType === 'cSupplier';
}
// --- Transformación: arma el payload para un destino puntual ---
// Solo se replican estos campos — el resto del payload de HQ01 se descarta.
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
// --- Mapeo: GroupCode vive en Postgres (scenario_value_mappings), no acá ---
const groupCode = record.GroupCode === undefined || record.GroupCode === null ? undefined : String(record.GroupCode);
const mappedGroupCode = groupCode === undefined ? undefined : ctx.mapValue('GroupCode', groupCode);
if (groupCode !== undefined && mappedGroupCode === undefined) {
throw new Error(`GroupCode '${groupCode}' no tiene mapeo cargado para el destino '${ctx.destinationCode}'`);
}
return {
CardCode: record.CardCode,
CardName: record.CardName,
CardType: record.CardType,
GroupCode: mappedGroupCode,
Phone1: record.Phone1,
Phone2: record.Phone2,
Notes: record.Notes,
FederalTaxID: record.FederalTaxID,
Currency: record.Currency,
DebitorAccount: record.DebitorAccount,
Valid: record.Valid,
Frozen: record.Frozen,
Properties1: record.Properties1,
};
}
+14
View File
@@ -0,0 +1,14 @@
// Escenario CHARTOFACCOUNTS: ChartOfAccounts, sin filtro, origen HQ01,
// destinos SUB_01/SUB_02 — ver docs/ESCENARIOS.md.
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: decide si este escenario aplica al registro ---
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean {
return true; // sin condición: aplica a todo el plan de cuentas
}
// --- Transformación: arma el payload para un destino puntual ---
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
// --- Mapeo: sin campos mapeados por ahora ---
return { ...record };
}
+41
View File
@@ -0,0 +1,41 @@
// Escenario JOURNALENTRIES: JournalEntries, destinationMode=payload, origen
// HQ01, destino resuelto por U_ITC_Target, keyStrategy=udfLink vía
// U_ITC_Source_JdtNum — ver docs/ESCENARIOS.md.
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: solo se replica si el asiento tiene una base destino asignada ---
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean {
const target = record.U_ITC_Target;
return typeof target === 'string' && target.trim() !== '';
}
// --- Transformación: arma el payload para el destino resuelto por U_ITC_Target ---
// Sin Series por ahora (las series no son portables entre compañías — se
// revisa más adelante cómo relacionarlas). AccountCode ya debería existir en
// el destino porque lo replica el escenario CHARTOFACCOUNTS.
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
const lines = Array.isArray(record.JournalEntryLines) ? record.JournalEntryLines : [];
const journalEntryLines = lines.map((line: Record<string, unknown>) => ({
AccountCode: line.AccountCode,
Debit: line.Credit,
Credit: line.Debit,
ShortName: line.ShortName,
LineMemo: line.LineMemo,
Reference1: line.Reference1,
Reference2: line.Reference2,
TaxDate: line.TaxDate,
ProjectCode: line.ProjectCode,
CostingCode: line.CostingCode,
}));
return {
ReferenceDate: record.ReferenceDate,
Memo: record.Memo,
Reference: record.Reference,
Reference2: record.Reference2,
TaxDate: record.TaxDate,
DueDate: record.DueDate,
JournalEntryLines: journalEntryLines,
};
}
+14
View File
@@ -0,0 +1,14 @@
// Escenario PROFITCENTERS: ProfitCenters, sin filtro, origen HQ01, destinos
// SUB_01/SUB_02/SUB_03 — ver docs/ESCENARIOS.md.
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: decide si este escenario aplica al registro ---
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean {
return true; // sin condición: aplica a todos los ProfitCenters
}
// --- Transformación: arma el payload para un destino puntual ---
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
// --- Mapeo: sin campos mapeados por ahora ---
return { ...record };
}
+14
View File
@@ -0,0 +1,14 @@
// Escenario SALESPERSONS: SalesPersons, sin filtro, origen HQ01, destinos
// SUB_01/SUB_03, keyStrategy=udfLink vía U_HQ_SlpCode — ver docs/ESCENARIOS.md.
import type { ScenarioFilterContext, ScenarioTransformContext } from '../scenarioTransform.ts';
// --- Filtro: decide si este escenario aplica al registro ---
export function filter(record: Record<string, unknown>, ctx: ScenarioFilterContext): boolean {
return true; // sin condición: aplica a todos los SalesPersons
}
// --- Transformación: arma el payload para un destino puntual ---
export function transform(record: Record<string, unknown>, ctx: ScenarioTransformContext): Record<string, unknown> {
// --- Mapeo: sin campos mapeados por ahora ---
return { ...record };
}
+56
View File
@@ -0,0 +1,56 @@
import axios, { type AxiosInstance } from 'axios';
import https from 'node:https';
import type { ServiceLayerTarget } from './config.ts';
// Cliente mínimo del Service Layer: Login + GET/POST/PATCH con la cookie
// de sesión. No cachea la sesión entre corridas del worker (cada ciclo
// vuelve a loguearse) para no lidiar con expiración de B1SESSION (~30 min).
export class ServiceLayerClient {
private readonly http: AxiosInstance;
private loggedIn = false;
constructor(private readonly target: ServiceLayerTarget) {
this.http = axios.create({
baseURL: target.baseUrl,
httpsAgent: new https.Agent({ rejectUnauthorized: false }),
validateStatus: () => true, // manejamos los status code nosotros, sin que axios tire excepción
});
}
async login(): Promise<void> {
const res = await this.http.post('/Login', {
CompanyDB: this.target.companyDb,
UserName: this.target.username,
Password: this.target.password,
});
if (res.status !== 200) {
throw new Error(
`Login falló en ${this.target.name} (${this.target.baseUrl}): HTTP ${res.status} ${JSON.stringify(res.data)}`,
);
}
const setCookie = res.headers['set-cookie'];
if (setCookie) {
this.http.defaults.headers.common['Cookie'] = setCookie.map((c) => c.split(';')[0]).join('; ');
}
this.loggedIn = true;
}
private async ensureLoggedIn(): Promise<void> {
if (!this.loggedIn) await this.login();
}
async get(path: string) {
await this.ensureLoggedIn();
return this.http.get(path);
}
async post(path: string, body: unknown) {
await this.ensureLoggedIn();
return this.http.post(path, body);
}
async patch(path: string, body: unknown) {
await this.ensureLoggedIn();
return this.http.patch(path, body);
}
}
+19
View File
@@ -0,0 +1,19 @@
@url=https://sap10:50000/b1s/v2
###
# @no-reject-unauthorized
POST {{url}}/Login
Content-Type : application/json
{
"CompanyDB":"SBODemoCL",
"UserName": "manager",
"Password":"mngr"
}
###
# @no-reject-unauthorized
GET {{url}}/BusinessPartners?$filter=CardType eq 'cSupplier'
###
+16
View File
@@ -0,0 +1,16 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"esModuleInterop": true,
"strict": true,
"skipLibCheck": true,
"resolveJsonModule": true,
"types": ["node"],
"allowImportingTsExtensions": true,
"noEmit": true
},
"include": ["src/**/*.ts"],
"exclude": ["node_modules", "src/generated"]
}