Compare commits
4
Commits
e1c8ee9015
...
v1.1.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
501bbad1d9 | ||
|
|
5a2e149825 | ||
|
|
9e85c4cee8 | ||
|
|
ec312d9461 |
No files matched your search
@@ -10,6 +10,10 @@ POLL_INTERVAL_SECONDS=30
|
|||||||
# Reintentos por suscripción antes de pasarla a State='X' (Exceptional)
|
# Reintentos por suscripción antes de pasarla a State='X' (Exceptional)
|
||||||
RETRY_LIMIT=5
|
RETRY_LIMIT=5
|
||||||
|
|
||||||
|
# Máximo de eventos por POST a una suscripción (un backlog grande se parte
|
||||||
|
# en varios envíos, para no superar el tamaño de body del receptor)
|
||||||
|
MAX_EVENTS_PER_BATCH=30
|
||||||
|
|
||||||
# El destino y la autenticación de cada envío NO se configuran acá: salen
|
# El destino y la autenticación de cada envío NO se configuran acá: salen
|
||||||
# de @OEVC (WebhookURL/AuthType/AuthCred), administrado con
|
# de @OEVC (WebhookURL/AuthType/AuthCred), administrado con
|
||||||
# `scripts/manage-subscriptions.ts` — una suscripción puede apuntar a
|
# `scripts/manage-subscriptions.ts` — una suscripción puede apuntar a
|
||||||
|
|||||||
@@ -3,3 +3,4 @@ node_modules
|
|||||||
.env
|
.env
|
||||||
logs/
|
logs/
|
||||||
tmp-test-events/
|
tmp-test-events/
|
||||||
|
.env.test
|
||||||
@@ -0,0 +1,60 @@
|
|||||||
|
# CLAUDE.md
|
||||||
|
|
||||||
|
Notas operativas para trabajar en este repo. El diseño completo está en
|
||||||
|
`docs/DISENO.md`.
|
||||||
|
|
||||||
|
## Entorno de este servidor
|
||||||
|
|
||||||
|
- El `.env` apunta a **producción** (`SBOPRDLIBROS`, SAP B1 sobre HANA).
|
||||||
|
Cualquier `npm run run-once` / `npm start` escribe en `@OEVN`/`@EVN1` y
|
||||||
|
envía webhooks reales a la suscripción activa en `@OEVC`.
|
||||||
|
- El dispatcher corre como servicio Windows NSSM `sap-legacy-webhook`
|
||||||
|
(cuenta LocalSystem). **Detener el servicio antes de correrlo a mano**
|
||||||
|
(`Stop-Service sap-legacy-webhook`) y volver a iniciarlo después: dos
|
||||||
|
procesos en paralelo comparten eventos y duplican entregas.
|
||||||
|
- Los logs diarios (`logs/dispatcher-YYYY-MM-DD.log`) los crea SYSTEM; un
|
||||||
|
usuario normal no puede escribirlos. El logger cae a solo-consola en ese
|
||||||
|
caso (no es un error).
|
||||||
|
|
||||||
|
## Lecciones aprendidas (2026-10-20)
|
||||||
|
|
||||||
|
- **Ticks solapados**: con backlog grande un tick dura más que
|
||||||
|
`POLL_INTERVAL_SECONDS`, y node-cron lanzaba otro en paralelo sobre la
|
||||||
|
misma sesión de Service Layer → `-2035 already exists` en `@EVN1` y
|
||||||
|
`-1116` (a veces "Unknown error", a veces violación de `KEVN1_NAME` con
|
||||||
|
la clave de *otro* request). Corregido con el guard `tickRunning` en
|
||||||
|
`src/index.ts`. Si reaparecen `-1116` cruzados, sospechar concurrencia
|
||||||
|
sobre la misma sesión antes que datos.
|
||||||
|
- **Tamaño de lote**: el receptor rechaza bodies grandes (HTTP 413). Los
|
||||||
|
envíos se parten en lotes de `MAX_EVENTS_PER_BATCH` (30 por defecto).
|
||||||
|
|
||||||
|
- **Eventos atascados en `P`**: si el PATCH final de un tick falla, el evento
|
||||||
|
queda en `P` con todas sus entregas en `D`, y el dispatcher no lo
|
||||||
|
recupera (solo recalcula eventos con entregas pendientes). Detectar con
|
||||||
|
`@OEVN` en `P` vs `@EVN1` en `F` = 0; se corrigieron 733 a mano el
|
||||||
|
2026-10-20.
|
||||||
|
|
||||||
|
## Versionado
|
||||||
|
|
||||||
|
SemVer en `package.json`, con tag `vX.Y.Z` en git por cada versión:
|
||||||
|
- **patch**: corrección sin cambio de comportamiento.
|
||||||
|
- **minor**: funcionalidad nueva compatible (ej. una variable nueva en `.env`).
|
||||||
|
- **major**: cambio incompatible (estructura de tablas, SP, formato del
|
||||||
|
CloudEvent).
|
||||||
|
|
||||||
|
Al cerrar un cambio: `npm version <patch|minor|major> --no-git-tag-version`,
|
||||||
|
commit con el código, `git tag -a vX.Y.Z -m "..."` y
|
||||||
|
`git push origin main --follow-tags`.
|
||||||
|
|
||||||
|
## Mejoras pendientes
|
||||||
|
|
||||||
|
- **Lock de instancia única**: archivo de bloqueo (ej. `logs/dispatcher.lock`
|
||||||
|
con el PID) para que una segunda instancia —servicio o ejecución manual—
|
||||||
|
se cierre al arrancar. Hoy solo lo evita la regla operativa de arriba.
|
||||||
|
- **Paralelizar llamadas al Service Layer** dentro de un lote (GET de
|
||||||
|
eventos / PATCH de `@EVN1` de a 5–10) para vaciar backlogs grandes más
|
||||||
|
rápido. Cuidado: misma sesión de SL — validar que tolera concurrencia o
|
||||||
|
usar sesiones separadas (ver lección de ticks solapados).
|
||||||
|
- **Causa del `-1116` aislado**: tras el fix de solapamiento no se volvió a
|
||||||
|
ver; si aparece con un solo tick en curso, el log ahora indica método,
|
||||||
|
ruta y body exactos.
|
||||||
@@ -100,15 +100,70 @@ npm run run-once # una sola pasada, útil para probar
|
|||||||
npm run start # corre en loop
|
npm run start # corre en loop
|
||||||
```
|
```
|
||||||
|
|
||||||
|
## Correr como servicio de Windows (NSSM)
|
||||||
|
|
||||||
|
En Windows Server, el dispatcher (`npm start`) se deja instalado como
|
||||||
|
servicio con [NSSM](https://nssm.cc/) para que arranque solo con el
|
||||||
|
server y se reinicie si el proceso muere. Pasos (reemplazar
|
||||||
|
`<NombreServicio>` y las rutas por las reales):
|
||||||
|
|
||||||
|
```powershell
|
||||||
|
# 1. Instalar el servicio: Application = node.exe, Arguments = CLI de tsx + entrypoint
|
||||||
|
nssm install <NombreServicio> "C:\Program Files\nodejs\node.exe" "C:\ruta\al\proyecto\node_modules\tsx\dist\cli.mjs src/index.ts"
|
||||||
|
|
||||||
|
# 2. Directorio de trabajo -- IMPRESCINDIBLE: acá se resuelve el .env (dotenv lee del cwd)
|
||||||
|
nssm set <NombreServicio> AppDirectory "C:\ruta\al\proyecto"
|
||||||
|
|
||||||
|
# 3. Arranque automático con el server
|
||||||
|
nssm set <NombreServicio> Start SERVICE_AUTO_START
|
||||||
|
|
||||||
|
# 4. Reinicio automático si el proceso crashea
|
||||||
|
nssm set <NombreServicio> AppExit Default Restart
|
||||||
|
|
||||||
|
# 5. Captura de stderr como red de seguridad (la app ya loguea sola, ver
|
||||||
|
# "Logs" abajo, así que Output/stdout no hace falta -- solo serviría
|
||||||
|
# para duplicar lo mismo)
|
||||||
|
nssm set <NombreServicio> AppStderr "C:\ruta\al\proyecto\logs\nssm-stderr.log"
|
||||||
|
|
||||||
|
# 6. Arrancarlo
|
||||||
|
nssm start <NombreServicio>
|
||||||
|
Get-Service <NombreServicio> # debe decir "Running"
|
||||||
|
```
|
||||||
|
|
||||||
|
`nssm install <NombreServicio>` sin más argumentos abre un instalador
|
||||||
|
gráfico equivalente (útil si hay sesión de escritorio): en la pestaña
|
||||||
|
*Application* van el Path/Startup directory/Arguments (pasos 1-2), en
|
||||||
|
*Details* el Startup type (paso 3), en *Exit actions* el restart (paso
|
||||||
|
4) y en *I/O* el stderr (paso 5).
|
||||||
|
|
||||||
|
⚠️ No usar `npm.cmd` como Application — NSSM no trackea bien el proceso
|
||||||
|
Node hijo que lanza npm, y al parar el servicio queda un proceso Node
|
||||||
|
huérfano corriendo.
|
||||||
|
|
||||||
|
Para probar que el servicio instalado esté vivo (más allá de que
|
||||||
|
`Get-Service` diga `Running`): seguir el log del día en vivo y confirmar
|
||||||
|
que aparecen ticks nuevos cada `POLL_INTERVAL_SECONDS`.
|
||||||
|
|
||||||
|
```powershell
|
||||||
|
Get-Content "C:\ruta\al\proyecto\logs\dispatcher-$(Get-Date -Format yyyy-MM-dd).log" -Tail 20 -Wait
|
||||||
|
```
|
||||||
|
|
||||||
|
Para desinstalar o reconfigurar: `nssm stop <NombreServicio>` seguido de
|
||||||
|
`nssm remove <NombreServicio> confirm`, o `nssm edit <NombreServicio>`
|
||||||
|
para el GUI de edición.
|
||||||
|
|
||||||
## Logs
|
## Logs
|
||||||
|
|
||||||
Cada tick escribe a `logs/dispatcher-YYYY-MM-DD.log` (uno por día) además
|
Cada tick escribe a `logs/dispatcher-YYYY-MM-DD.log` (uno por día,
|
||||||
de la consola — línea simple `<timestamp ISO> [NIVEL] mensaje`, legible
|
rotando a medianoche hora local del server) además de la consola — línea
|
||||||
por cualquier visualizador genérico (`tail -f logs/dispatcher-*.log`,
|
simple `<timestamp ISO con offset local> [NIVEL] mensaje` (ej.
|
||||||
`lnav`, etc.). Siempre deja al menos una línea de resumen por tick
|
`2026-09-16T13:50:32.565-03:00 [INFO] ...`), legible por cualquier
|
||||||
(`Tick: N evento(s) nuevo(s), M pendiente(s) de entrega`), incluso sin
|
visualizador genérico (`tail -f logs/dispatcher-*.log`, `lnav`, etc.). Se
|
||||||
actividad — para poder confirmar que el proceso sigue vivo. No versionado
|
usa hora local (no UTC) a propósito para que los logs sean auditables
|
||||||
(`.gitignore`).
|
directo contra la hora de pared, sin restar husos horarios a mano.
|
||||||
|
Siempre deja al menos una línea de resumen por tick (`Tick: N evento(s)
|
||||||
|
nuevo(s), M pendiente(s) de entrega`), incluso sin actividad — para poder
|
||||||
|
confirmar que el proceso sigue vivo. No versionado (`.gitignore`).
|
||||||
|
|
||||||
## Estructura
|
## Estructura
|
||||||
|
|
||||||
|
|||||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
|||||||
{
|
{
|
||||||
"name": "sap-legacy-webhook-emulator",
|
"name": "sap-legacy-webhook-emulator",
|
||||||
"version": "1.0.0",
|
"version": "1.1.0",
|
||||||
"lockfileVersion": 3,
|
"lockfileVersion": 3,
|
||||||
"requires": true,
|
"requires": true,
|
||||||
"packages": {
|
"packages": {
|
||||||
"": {
|
"": {
|
||||||
"name": "sap-legacy-webhook-emulator",
|
"name": "sap-legacy-webhook-emulator",
|
||||||
"version": "1.0.0",
|
"version": "1.1.0",
|
||||||
"license": "UNLICENSED",
|
"license": "UNLICENSED",
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"axios": "^1.19.0",
|
"axios": "^1.19.0",
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "sap-legacy-webhook-emulator",
|
"name": "sap-legacy-webhook-emulator",
|
||||||
"version": "1.0.0",
|
"version": "1.1.0",
|
||||||
"description": "Emula el Webhook Messenger nativo de SAP B1 (FP2605+) para clientes en versiones anteriores",
|
"description": "Emula el Webhook Messenger nativo de SAP B1 (FP2605+) para clientes en versiones anteriores",
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"main": "src/index.ts",
|
"main": "src/index.ts",
|
||||||
|
|||||||
@@ -8,6 +8,7 @@
|
|||||||
// npx tsx scripts/manage-subscriptions.ts subscription:update-url <webhookId> <webhookUrl>
|
// npx tsx scripts/manage-subscriptions.ts subscription:update-url <webhookId> <webhookUrl>
|
||||||
// npx tsx scripts/manage-subscriptions.ts subscription:pause <webhookId>
|
// npx tsx scripts/manage-subscriptions.ts subscription:pause <webhookId>
|
||||||
// npx tsx scripts/manage-subscriptions.ts subscription:resume <webhookId>
|
// npx tsx scripts/manage-subscriptions.ts subscription:resume <webhookId>
|
||||||
|
// npx tsx scripts/manage-subscriptions.ts subscription:delete <webhookId>
|
||||||
//
|
//
|
||||||
// npx tsx scripts/manage-subscriptions.ts filter:add <webhookId> <bizObject> <transType> <objectType>
|
// npx tsx scripts/manage-subscriptions.ts filter:add <webhookId> <bizObject> <transType> <objectType>
|
||||||
// npx tsx scripts/manage-subscriptions.ts filter:remove <webhookId> <bizObject> <transType>
|
// npx tsx scripts/manage-subscriptions.ts filter:remove <webhookId> <bizObject> <transType>
|
||||||
@@ -15,6 +16,8 @@
|
|||||||
|
|
||||||
import { config } from '../src/config.ts';
|
import { config } from '../src/config.ts';
|
||||||
import { ServiceLayerClient } from '../src/service-layer-client.ts';
|
import { ServiceLayerClient } from '../src/service-layer-client.ts';
|
||||||
|
import * as evn1 from '../src/dispatcher/evn1-repository.ts';
|
||||||
|
import * as oevn from '../src/dispatcher/oevn-repository.ts';
|
||||||
import { generateCode } from '../src/lib/codes.ts';
|
import { generateCode } from '../src/lib/codes.ts';
|
||||||
import { userTableEntity } from '../src/lib/service-layer-paths.ts';
|
import { userTableEntity } from '../src/lib/service-layer-paths.ts';
|
||||||
|
|
||||||
@@ -87,7 +90,43 @@ async function main() {
|
|||||||
const [webhookId] = args;
|
const [webhookId] = args;
|
||||||
if (!webhookId) fail('subscription:resume <webhookId>');
|
if (!webhookId) fail('subscription:resume <webhookId>');
|
||||||
await client.patch(`/${OEVC}('${webhookId}')`, { U_State: 'A' });
|
await client.patch(`/${OEVC}('${webhookId}')`, { U_State: 'A' });
|
||||||
console.log(`Suscripción ${webhookId} reactivada (State='A').`);
|
const reset = await evn1.resetRetries(client, webhookId);
|
||||||
|
console.log(`Suscripción ${webhookId} reactivada (State='A'); ${reset} entrega(s) con U_Retry reiniciado a 0.`);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
case 'subscription:delete': {
|
||||||
|
const [webhookId] = args;
|
||||||
|
if (!webhookId) fail('subscription:delete <webhookId>');
|
||||||
|
|
||||||
|
// 1) Entregas fallidas/pendientes (Status='F') de esta suscripción. Las ya
|
||||||
|
// entregadas (Status='D') se conservan como historial.
|
||||||
|
const failed = await evn1.listPendingForWebhook(client, webhookId);
|
||||||
|
for (const row of failed) await evn1.deleteRow(client, row.code);
|
||||||
|
|
||||||
|
// 2) Eventos de @OEVN que quedaron sin ninguna entrega asociada (solo
|
||||||
|
// existían para esta suscripción) se borran; los demás se recalculan.
|
||||||
|
let eventsDeleted = 0;
|
||||||
|
for (const eventId of new Set(failed.map((r) => r.eventId))) {
|
||||||
|
const remaining = await evn1.listForEvent(client, eventId);
|
||||||
|
if (remaining.length === 0) {
|
||||||
|
await oevn.deleteEvent(client, eventId);
|
||||||
|
eventsDeleted++;
|
||||||
|
} else {
|
||||||
|
await oevn.setEventStatus(client, eventId, remaining.every((r) => r.status === 'D') ? 'D' : 'P');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 3) Filtros de @EVC1.
|
||||||
|
const filters = await client.getAll(`/${EVC1}?$filter=${encodeURIComponent(`U_WebhookID eq '${webhookId}'`)}`);
|
||||||
|
for (const f of filters) await client.delete(`/${EVC1}('${f.Code}')`);
|
||||||
|
|
||||||
|
// 4) La suscripción en sí.
|
||||||
|
await client.delete(`/${OEVC}('${webhookId}')`);
|
||||||
|
console.log(
|
||||||
|
`Suscripción ${webhookId} eliminada: ${filters.length} filtro(s), ${failed.length} entrega(s) fallida(s), ` +
|
||||||
|
`${eventsDeleted} evento(s) huérfano(s) de @OEVN.`,
|
||||||
|
);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -140,7 +179,7 @@ async function main() {
|
|||||||
default:
|
default:
|
||||||
fail(
|
fail(
|
||||||
`comando desconocido: ${command ?? '(ninguno)'}\nComandos disponibles: subscription:add, subscription:list, ` +
|
`comando desconocido: ${command ?? '(ninguno)'}\nComandos disponibles: subscription:add, subscription:list, ` +
|
||||||
'subscription:update-url, subscription:pause, subscription:resume, filter:add, filter:remove, filter:list',
|
'subscription:update-url, subscription:pause, subscription:resume, subscription:delete, filter:add, filter:remove, filter:list',
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,6 +19,9 @@ export const config = {
|
|||||||
// cualquier URL, no solo a una instancia fija de sap-webhook-api.
|
// cualquier URL, no solo a una instancia fija de sap-webhook-api.
|
||||||
pollIntervalSeconds: Number(process.env.POLL_INTERVAL_SECONDS ?? 30),
|
pollIntervalSeconds: Number(process.env.POLL_INTERVAL_SECONDS ?? 30),
|
||||||
retryLimit: Number(process.env.RETRY_LIMIT ?? 5),
|
retryLimit: Number(process.env.RETRY_LIMIT ?? 5),
|
||||||
|
// Tope de eventos por POST a una suscripción — sin tope, un backlog grande
|
||||||
|
// se manda entero en un solo body y el receptor lo rechaza (HTTP 413).
|
||||||
|
maxEventsPerBatch: Number(process.env.MAX_EVENTS_PER_BATCH ?? 30),
|
||||||
};
|
};
|
||||||
|
|
||||||
function requireEnv(name: string): string {
|
function requireEnv(name: string): string {
|
||||||
|
|||||||
@@ -26,6 +26,10 @@ function toRow(raw: Record<string, unknown>): Evn1Row {
|
|||||||
export async function findExisting(client: ServiceLayerClient, eventId: string, webhookId: string): Promise<Evn1Row | null> {
|
export async function findExisting(client: ServiceLayerClient, eventId: string, webhookId: string): Promise<Evn1Row | null> {
|
||||||
const filter = encodeURIComponent(`U_EventID eq '${eventId}' and U_WebhookID eq '${webhookId}'`);
|
const filter = encodeURIComponent(`U_EventID eq '${eventId}' and U_WebhookID eq '${webhookId}'`);
|
||||||
const res = await client.get(`/${ENTITY}?$filter=${filter}`);
|
const res = await client.get(`/${ENTITY}?$filter=${filter}`);
|
||||||
|
// Un GET fallido no es "no existe": tratarlo así lleva a un POST duplicado.
|
||||||
|
if (res.status !== 200) {
|
||||||
|
throw new Error(`Service Layer respondió HTTP ${res.status} buscando @EVN1 de ${eventId}: ${JSON.stringify(res.data)}`);
|
||||||
|
}
|
||||||
const rows = (res.data?.value as Record<string, unknown>[]) ?? [];
|
const rows = (res.data?.value as Record<string, unknown>[]) ?? [];
|
||||||
return rows.length > 0 ? toRow(rows[0]) : null;
|
return rows.length > 0 ? toRow(rows[0]) : null;
|
||||||
}
|
}
|
||||||
@@ -70,3 +74,24 @@ export async function markFailedAttempt(client: ServiceLayerClient, row: Evn1Row
|
|||||||
U_LastSentTM: hhmmss(now),
|
U_LastSentTM: hhmmss(now),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Pone en 0 el contador de reintentos de las entregas pendientes de una
|
||||||
|
// suscripción. Se usa al reactivarla (State X -> A): sin esto, las filas
|
||||||
|
// siguen con Retry >= RETRY_LIMIT y la suscripción vuelve a X en el
|
||||||
|
// siguiente tick sin intentar enviar nada.
|
||||||
|
export async function resetRetries(client: ServiceLayerClient, webhookId: string): Promise<number> {
|
||||||
|
const filter = encodeURIComponent(`U_WebhookID eq '${webhookId}' and U_Status eq 'F'`);
|
||||||
|
const rows = (await client.getAll(`/${ENTITY}?$filter=${filter}`)).map(toRow).filter((r) => r.retry > 0);
|
||||||
|
for (const row of rows) await client.patch(`/${ENTITY}('${row.code}')`, { U_Retry: 0 });
|
||||||
|
return rows.length;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Entregas fallidas/pendientes (Status='F') de una suscripción.
|
||||||
|
export async function listPendingForWebhook(client: ServiceLayerClient, webhookId: string): Promise<Evn1Row[]> {
|
||||||
|
const filter = encodeURIComponent(`U_WebhookID eq '${webhookId}' and U_Status eq 'F'`);
|
||||||
|
return (await client.getAll(`/${ENTITY}?$filter=${filter}`)).map(toRow);
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function deleteRow(client: ServiceLayerClient, code: string): Promise<void> {
|
||||||
|
await client.delete(`/${ENTITY}('${code}')`);
|
||||||
|
}
|
||||||
@@ -27,38 +27,68 @@ async function registerNewEvents(client: ServiceLayerClient): Promise<number> {
|
|||||||
if (newEvents.length === 0) return 0;
|
if (newEvents.length === 0) return 0;
|
||||||
|
|
||||||
const filters = await subscriptions.listFilters(client);
|
const filters = await subscriptions.listFilters(client);
|
||||||
|
logger.info(`[dispatcher] Registrando ${newEvents.length} evento(s) nuevo(s) (${filters.length} filtro(s) de suscripción).`);
|
||||||
|
|
||||||
for (const event of newEvents) {
|
let failed = 0;
|
||||||
// Eco de una escritura nuestra sobre @OEVC/@EVC1/@OEVN/@EVN1 (ver
|
for (const [index, event] of newEvents.entries()) {
|
||||||
// own-table-guard.ts) — nunca se toca, ni se resuelve ni se borra,
|
// Con miles de eventos por tick, sin un rastro intermedio el proceso
|
||||||
// para no perpetuar la cascada.
|
// parece colgado hasta el log final.
|
||||||
if (isEchoOfOwnTable(event.objectType)) continue;
|
if (index > 0 && index % PROGRESS_EVERY === 0) {
|
||||||
|
logger.info(`[dispatcher] Registro: ${index}/${newEvents.length} evento(s) procesado(s), ${failed} con error.`);
|
||||||
let bizObject: string;
|
}
|
||||||
|
// Un evento que falla no corta el tick: se registra con su contexto y
|
||||||
|
// queda en Status='N' para reintentarse en el próximo tick, sin
|
||||||
|
// bloquear a todos los que vienen detrás.
|
||||||
try {
|
try {
|
||||||
bizObject = bizObjectFor(event.objectType);
|
await registerEvent(client, event, filters);
|
||||||
} catch {
|
} catch (err) {
|
||||||
logger.warn(`[dispatcher] ObjectType desconocido '${event.objectType}' (evento ${event.code}) — se descarta.`);
|
failed++;
|
||||||
await oevn.deleteEvent(client, event.code);
|
logger.error(
|
||||||
continue;
|
`[dispatcher] Error registrando evento ${event.code} (ObjectType='${event.objectType}', TransType='${event.transType}', ` +
|
||||||
|
`${event.fieldNames}=${event.fieldValue}):`,
|
||||||
|
err,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
const matches = filters.filter((f) => f.bizObject === bizObject && f.transType === event.transType);
|
|
||||||
if (matches.length === 0) {
|
|
||||||
logger.info(`[dispatcher] Sin suscripciones para ${bizObject}/${event.transType} (evento ${event.code}) — se descarta.`);
|
|
||||||
await oevn.deleteEvent(client, event.code);
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
for (const match of matches) {
|
|
||||||
const existing = await evn1.findExisting(client, event.code, match.webhookId);
|
|
||||||
if (!existing) await evn1.create(client, event.code, match.webhookId);
|
|
||||||
}
|
|
||||||
await oevn.setEventStatus(client, event.code, 'P');
|
|
||||||
}
|
}
|
||||||
|
if (failed > 0) logger.warn(`[dispatcher] Registro: ${failed}/${newEvents.length} evento(s) con error, quedan para el próximo tick.`);
|
||||||
return newEvents.length;
|
return newEvents.length;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const PROGRESS_EVERY = 100;
|
||||||
|
|
||||||
|
async function registerEvent(
|
||||||
|
client: ServiceLayerClient,
|
||||||
|
event: oevn.OevnRow,
|
||||||
|
filters: Awaited<ReturnType<typeof subscriptions.listFilters>>,
|
||||||
|
): Promise<void> {
|
||||||
|
// Eco de una escritura nuestra sobre @OEVC/@EVC1/@OEVN/@EVN1 (ver
|
||||||
|
// own-table-guard.ts) — nunca se toca, ni se resuelve ni se borra,
|
||||||
|
// para no perpetuar la cascada.
|
||||||
|
if (isEchoOfOwnTable(event.objectType)) return;
|
||||||
|
|
||||||
|
let bizObject: string;
|
||||||
|
try {
|
||||||
|
bizObject = bizObjectFor(event.objectType);
|
||||||
|
} catch {
|
||||||
|
logger.warn(`[dispatcher] ObjectType desconocido '${event.objectType}' (evento ${event.code}) — se descarta.`);
|
||||||
|
await oevn.deleteEvent(client, event.code);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const matches = filters.filter((f) => f.bizObject === bizObject && f.transType === event.transType);
|
||||||
|
if (matches.length === 0) {
|
||||||
|
logger.info(`[dispatcher] Sin suscripciones para ${bizObject}/${event.transType} (evento ${event.code}) — se descarta.`);
|
||||||
|
await oevn.deleteEvent(client, event.code);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
for (const match of matches) {
|
||||||
|
const existing = await evn1.findExisting(client, event.code, match.webhookId);
|
||||||
|
if (!existing) await evn1.create(client, event.code, match.webhookId);
|
||||||
|
}
|
||||||
|
await oevn.setEventStatus(client, event.code, 'P');
|
||||||
|
}
|
||||||
|
|
||||||
async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number> {
|
async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number> {
|
||||||
const pending = await evn1.listPending(client);
|
const pending = await evn1.listPending(client);
|
||||||
if (pending.length === 0) return 0;
|
if (pending.length === 0) return 0;
|
||||||
@@ -82,8 +112,16 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
|
|||||||
byWebhookId.set(row.webhookId, group);
|
byWebhookId.set(row.webhookId, group);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
logger.info(`[dispatcher] Entrega: ${pending.length} fila(s) pendiente(s) en @EVN1, ${byWebhookId.size} suscripción(es) a despachar.`);
|
||||||
for (const [webhookId, rows] of byWebhookId) {
|
for (const [webhookId, rows] of byWebhookId) {
|
||||||
await sendGroup(client, activeSubscriptions.get(webhookId)!, rows);
|
const batchSize = config.maxEventsPerBatch;
|
||||||
|
const batchCount = Math.ceil(rows.length / batchSize);
|
||||||
|
logger.info(`[dispatcher] Preparando ${rows.length} evento(s) para ${webhookId} en ${batchCount} lote(s) de hasta ${batchSize}.`);
|
||||||
|
// Cada lote se marca entregado/fallido por separado: uno rechazado no
|
||||||
|
// arrastra a los demás.
|
||||||
|
for (let i = 0; i < rows.length; i += batchSize) {
|
||||||
|
await sendGroup(client, activeSubscriptions.get(webhookId)!, rows.slice(i, i + batchSize));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
await recomputeEventStatuses(client, [...new Set(pending.map((r) => r.eventId))]);
|
await recomputeEventStatuses(client, [...new Set(pending.map((r) => r.eventId))]);
|
||||||
@@ -135,7 +173,11 @@ async function sendGroup(
|
|||||||
}
|
}
|
||||||
|
|
||||||
async function recomputeEventStatuses(client: ServiceLayerClient, eventIds: string[]): Promise<void> {
|
async function recomputeEventStatuses(client: ServiceLayerClient, eventIds: string[]): Promise<void> {
|
||||||
for (const eventId of eventIds) {
|
logger.info(`[dispatcher] Recalculando Status de ${eventIds.length} evento(s) en @OEVN.`);
|
||||||
|
for (const [index, eventId] of eventIds.entries()) {
|
||||||
|
if (index > 0 && index % PROGRESS_EVERY === 0) {
|
||||||
|
logger.info(`[dispatcher] Recalculo: ${index}/${eventIds.length} evento(s).`);
|
||||||
|
}
|
||||||
const rows = await evn1.listForEvent(client, eventId);
|
const rows = await evn1.listForEvent(client, eventId);
|
||||||
if (rows.length === 0) continue;
|
if (rows.length === 0) continue;
|
||||||
const allDelivered = rows.every((r) => r.status === 'D');
|
const allDelivered = rows.every((r) => r.status === 'D');
|
||||||
|
|||||||
@@ -6,11 +6,24 @@ import { logger } from './lib/logger.ts';
|
|||||||
|
|
||||||
const client = new ServiceLayerClient(config.serviceLayer);
|
const client = new ServiceLayerClient(config.serviceLayer);
|
||||||
|
|
||||||
|
// Un tick con backlog grande dura más que POLL_INTERVAL_SECONDS: sin este
|
||||||
|
// guard, cron lanza otro en paralelo sobre los mismos eventos y la misma
|
||||||
|
// sesión de Service Layer (duplicados -2035 y errores -1116 cruzados entre
|
||||||
|
// requests concurrentes).
|
||||||
|
let tickRunning = false;
|
||||||
|
|
||||||
async function runOnce(): Promise<void> {
|
async function runOnce(): Promise<void> {
|
||||||
|
if (tickRunning) {
|
||||||
|
logger.info('[dispatcher] El tick anterior sigue en curso — se omite este.');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
tickRunning = true;
|
||||||
try {
|
try {
|
||||||
await processTick(client);
|
await processTick(client);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
logger.error('[dispatcher] Error procesando el tick:', err);
|
logger.error('[dispatcher] Error procesando el tick:', err);
|
||||||
|
} finally {
|
||||||
|
tickRunning = false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+75
-5
@@ -1,5 +1,6 @@
|
|||||||
import { appendFileSync, mkdirSync } from 'node:fs';
|
import { appendFileSync, mkdirSync } from 'node:fs';
|
||||||
import { fileURLToPath } from 'node:url';
|
import { fileURLToPath } from 'node:url';
|
||||||
|
import { inspect } from 'node:util';
|
||||||
|
|
||||||
// Log a archivo, uno por día (logs/dispatcher-YYYY-MM-DD.log), además de
|
// Log a archivo, uno por día (logs/dispatcher-YYYY-MM-DD.log), además de
|
||||||
// consola — para que un servicio Windows (NSSM/PM2/Task Scheduler, ver
|
// consola — para que un servicio Windows (NSSM/PM2/Task Scheduler, ver
|
||||||
@@ -9,14 +10,76 @@ import { fileURLToPath } from 'node:url';
|
|||||||
const LOGS_DIR = fileURLToPath(new URL('../../logs/', import.meta.url));
|
const LOGS_DIR = fileURLToPath(new URL('../../logs/', import.meta.url));
|
||||||
mkdirSync(LOGS_DIR, { recursive: true });
|
mkdirSync(LOGS_DIR, { recursive: true });
|
||||||
|
|
||||||
|
// Hora local del server (no UTC) con offset explícito, para que los logs
|
||||||
|
// sean auditables directamente contra la hora de pared sin tener que
|
||||||
|
// restar el huso horario a mano.
|
||||||
|
function pad(value: number, length = 2): string {
|
||||||
|
return String(value).padStart(length, '0');
|
||||||
|
}
|
||||||
|
|
||||||
|
function localTimestamp(date: Date): string {
|
||||||
|
const offsetMinutes = -date.getTimezoneOffset();
|
||||||
|
const sign = offsetMinutes >= 0 ? '+' : '-';
|
||||||
|
const absMinutes = Math.abs(offsetMinutes);
|
||||||
|
const offset = `${sign}${pad(Math.floor(absMinutes / 60))}:${pad(absMinutes % 60)}`;
|
||||||
|
return (
|
||||||
|
`${date.getFullYear()}-${pad(date.getMonth() + 1)}-${pad(date.getDate())}` +
|
||||||
|
`T${pad(date.getHours())}:${pad(date.getMinutes())}:${pad(date.getSeconds())}.${pad(date.getMilliseconds(), 3)}` +
|
||||||
|
offset
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
function logFilePath(date: Date): string {
|
function logFilePath(date: Date): string {
|
||||||
const day = date.toISOString().slice(0, 10);
|
const day = localTimestamp(date).slice(0, 10);
|
||||||
return `${LOGS_DIR}dispatcher-${day}.log`;
|
return `${LOGS_DIR}dispatcher-${day}.log`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// No poder escribir el archivo de log (ej. creado por el servicio como
|
||||||
|
// SYSTEM y corrido a mano con otro usuario: EPERM) nunca debe tumbar el
|
||||||
|
// dispatcher a mitad de un tick — se avisa una vez y se sigue solo por consola.
|
||||||
|
let fileWriteFailed = false;
|
||||||
|
|
||||||
function write(level: string, message: string): void {
|
function write(level: string, message: string): void {
|
||||||
|
if (fileWriteFailed) return;
|
||||||
const now = new Date();
|
const now = new Date();
|
||||||
appendFileSync(logFilePath(now), `${now.toISOString()} [${level}] ${message}\n`);
|
try {
|
||||||
|
appendFileSync(logFilePath(now), `${localTimestamp(now)} [${level}] ${message}\n`);
|
||||||
|
} catch (err) {
|
||||||
|
fileWriteFailed = true;
|
||||||
|
console.warn(`[logger] No se pudo escribir ${logFilePath(now)} (${err instanceof Error ? err.message : String(err)}) — se sigue solo por consola.`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Errores de red de axios (ECONNREFUSED, timeout, DNS...): primero se
|
||||||
|
// muestra una frase legible para el operador y después el error completo.
|
||||||
|
// El error de axios trae adjunto el body del Login, así que antes de
|
||||||
|
// imprimirlo se enmascara la contraseña.
|
||||||
|
const NETWORK_ERRORS: Record<string, string> = {
|
||||||
|
ECONNREFUSED: 'conexión rechazada (¿Service Layer detenido o puerto bloqueado?)',
|
||||||
|
ETIMEDOUT: 'tiempo de espera agotado',
|
||||||
|
ECONNABORTED: 'tiempo de espera agotado',
|
||||||
|
ENOTFOUND: 'no se pudo resolver el nombre del servidor',
|
||||||
|
ECONNRESET: 'el servidor cerró la conexión',
|
||||||
|
EHOSTUNREACH: 'servidor inalcanzable',
|
||||||
|
};
|
||||||
|
|
||||||
|
function describeError(err: unknown): string {
|
||||||
|
if (!(err instanceof Error)) return err !== undefined ? String(err) : '';
|
||||||
|
const axiosErr = err as Error & { isAxiosError?: boolean; code?: string; config?: { baseURL?: string; url?: string } };
|
||||||
|
if (axiosErr.isAxiosError) {
|
||||||
|
const target = `${axiosErr.config?.baseURL ?? ''}${axiosErr.config?.url ?? ''}`;
|
||||||
|
const reason = (axiosErr.code && NETWORK_ERRORS[axiosErr.code]) ?? axiosErr.message;
|
||||||
|
return `No se pudo conectar a ${target}: ${reason}`;
|
||||||
|
}
|
||||||
|
return err.message;
|
||||||
|
}
|
||||||
|
|
||||||
|
function redactPassword(err: unknown): unknown {
|
||||||
|
const config = (err as { config?: { data?: unknown } } | undefined)?.config;
|
||||||
|
if (config && typeof config.data === 'string') {
|
||||||
|
config.data = config.data.replace(/("Password"\s*:\s*")(?:[^"\\]|\\.)*"/g, '$1***"');
|
||||||
|
}
|
||||||
|
return err;
|
||||||
}
|
}
|
||||||
|
|
||||||
export const logger = {
|
export const logger = {
|
||||||
@@ -29,8 +92,15 @@ export const logger = {
|
|||||||
write('WARN', message);
|
write('WARN', message);
|
||||||
},
|
},
|
||||||
error(message: string, err?: unknown): void {
|
error(message: string, err?: unknown): void {
|
||||||
const detail = err instanceof Error ? (err.stack ?? err.message) : err !== undefined ? String(err) : '';
|
const summary = describeError(err);
|
||||||
console.error(message, err ?? '');
|
const headline = summary ? `${message} ${summary}` : message;
|
||||||
write('ERROR', detail ? `${message} ${detail}` : message);
|
if (err === undefined) {
|
||||||
|
console.error(headline);
|
||||||
|
write('ERROR', headline);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
const full = inspect(redactPassword(err));
|
||||||
|
console.error(`${headline}\n--- Detalle completo ---\n${full}`);
|
||||||
|
write('ERROR', `${headline}\n${full}`);
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
@@ -83,7 +83,14 @@ export class ServiceLayerClient {
|
|||||||
// una fila que "parece" creada/actualizada pero no lo está.
|
// una fila que "parece" creada/actualizada pero no lo está.
|
||||||
private assertOk(res: AxiosResponse): AxiosResponse {
|
private assertOk(res: AxiosResponse): AxiosResponse {
|
||||||
if (res.status < 200 || res.status >= 300) {
|
if (res.status < 200 || res.status >= 300) {
|
||||||
throw new Error(`Service Layer respondió HTTP ${res.status}: ${JSON.stringify(res.data)}`);
|
// Método + ruta + body enviado: sin esto el error no dice sobre qué
|
||||||
|
// fila/entidad falló, y con miles de eventos por tick no hay forma de
|
||||||
|
// ubicarlo.
|
||||||
|
const method = res.config.method?.toUpperCase() ?? '?';
|
||||||
|
throw new Error(
|
||||||
|
`Service Layer respondió HTTP ${res.status} en ${method} ${res.config.url}` +
|
||||||
|
` (body: ${res.config.data ?? '-'}): ${JSON.stringify(res.data)}`,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
return res;
|
return res;
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user