Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
501bbad1d9 | ||
|
|
5a2e149825 |
No files matched your search
@@ -10,6 +10,10 @@ POLL_INTERVAL_SECONDS=30
|
||||
# Reintentos por suscripción antes de pasarla a State='X' (Exceptional)
|
||||
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
|
||||
# de @OEVC (WebhookURL/AuthType/AuthCred), administrado con
|
||||
# `scripts/manage-subscriptions.ts` — una suscripción puede apuntar a
|
||||
|
||||
@@ -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.
|
||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "sap-legacy-webhook-emulator",
|
||||
"version": "1.0.0",
|
||||
"version": "1.1.0",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "sap-legacy-webhook-emulator",
|
||||
"version": "1.0.0",
|
||||
"version": "1.1.0",
|
||||
"license": "UNLICENSED",
|
||||
"dependencies": {
|
||||
"axios": "^1.19.0",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"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",
|
||||
"type": "module",
|
||||
"main": "src/index.ts",
|
||||
|
||||
@@ -19,6 +19,9 @@ export const config = {
|
||||
// cualquier URL, no solo a una instancia fija de sap-webhook-api.
|
||||
pollIntervalSeconds: Number(process.env.POLL_INTERVAL_SECONDS ?? 30),
|
||||
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 {
|
||||
|
||||
@@ -26,6 +26,10 @@ function toRow(raw: Record<string, unknown>): Evn1Row {
|
||||
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 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>[]) ?? [];
|
||||
return rows.length > 0 ? toRow(rows[0]) : null;
|
||||
}
|
||||
|
||||
@@ -27,42 +27,67 @@ async function registerNewEvents(client: ServiceLayerClient): Promise<number> {
|
||||
if (newEvents.length === 0) return 0;
|
||||
|
||||
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) {
|
||||
// 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)) continue;
|
||||
|
||||
let bizObject: string;
|
||||
let failed = 0;
|
||||
for (const [index, event] of newEvents.entries()) {
|
||||
// Con miles de eventos por tick, sin un rastro intermedio el proceso
|
||||
// parece colgado hasta el log final.
|
||||
if (index > 0 && index % PROGRESS_EVERY === 0) {
|
||||
logger.info(`[dispatcher] Registro: ${index}/${newEvents.length} evento(s) procesado(s), ${failed} con error.`);
|
||||
}
|
||||
// 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 {
|
||||
bizObject = bizObjectFor(event.objectType);
|
||||
} catch {
|
||||
logger.warn(`[dispatcher] ObjectType desconocido '${event.objectType}' (evento ${event.code}) — se descarta.`);
|
||||
await oevn.deleteEvent(client, event.code);
|
||||
continue;
|
||||
await registerEvent(client, event, filters);
|
||||
} catch (err) {
|
||||
failed++;
|
||||
logger.error(
|
||||
`[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;
|
||||
}
|
||||
|
||||
// Suscripciones que este proceso pasó a 'X'. Si después aparecen otra vez
|
||||
// activas (alguien las reactivó, por script o directo en Service Layer),
|
||||
// se reinicia U_Retry de sus entregas pendientes para darles otra tanda.
|
||||
const markedExceptional = new Set<string>();
|
||||
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> {
|
||||
const pending = await evn1.listPending(client);
|
||||
@@ -71,20 +96,11 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
|
||||
const activeSubscriptions = await subscriptions.listActiveSubscriptions(client);
|
||||
const byWebhookId = new Map<string, evn1.Evn1Row[]>();
|
||||
|
||||
for (const webhookId of [...markedExceptional]) {
|
||||
if (!activeSubscriptions.has(webhookId)) continue;
|
||||
markedExceptional.delete(webhookId);
|
||||
const reset = await evn1.resetRetries(client, webhookId);
|
||||
logger.info(`[dispatcher] Suscripción ${webhookId} reactivada — U_Retry reiniciado a 0 en ${reset} entrega(s).`);
|
||||
for (const row of pending) if (row.webhookId === webhookId) row.retry = 0;
|
||||
}
|
||||
|
||||
for (const row of pending) {
|
||||
if (row.retry >= config.retryLimit) {
|
||||
if (activeSubscriptions.has(row.webhookId)) {
|
||||
logger.warn(`[dispatcher] Suscripción ${row.webhookId} superó ${config.retryLimit} reintentos — pasa a State='X'.`);
|
||||
await subscriptions.markExceptional(client, row.webhookId);
|
||||
markedExceptional.add(row.webhookId);
|
||||
activeSubscriptions.delete(row.webhookId);
|
||||
}
|
||||
continue;
|
||||
@@ -96,11 +112,16 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
|
||||
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) {
|
||||
// Pudo pasar a 'X' después de armar el grupo (otra fila superó el límite).
|
||||
const subscription = activeSubscriptions.get(webhookId);
|
||||
if (!subscription) continue;
|
||||
await sendGroup(client, subscription, 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))]);
|
||||
@@ -152,7 +173,11 @@ async function sendGroup(
|
||||
}
|
||||
|
||||
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);
|
||||
if (rows.length === 0) continue;
|
||||
const allDelivered = rows.every((r) => r.status === 'D');
|
||||
|
||||
@@ -6,11 +6,24 @@ import { logger } from './lib/logger.ts';
|
||||
|
||||
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> {
|
||||
if (tickRunning) {
|
||||
logger.info('[dispatcher] El tick anterior sigue en curso — se omite este.');
|
||||
return;
|
||||
}
|
||||
tickRunning = true;
|
||||
try {
|
||||
await processTick(client);
|
||||
} catch (err) {
|
||||
logger.error('[dispatcher] Error procesando el tick:', err);
|
||||
} finally {
|
||||
tickRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+12
-1
@@ -34,9 +34,20 @@ function logFilePath(date: Date): string {
|
||||
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 {
|
||||
if (fileWriteFailed) return;
|
||||
const now = new Date();
|
||||
appendFileSync(logFilePath(now), `${localTimestamp(now)} [${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
|
||||
|
||||
@@ -83,7 +83,14 @@ export class ServiceLayerClient {
|
||||
// una fila que "parece" creada/actualizada pero no lo está.
|
||||
private assertOk(res: AxiosResponse): AxiosResponse {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user