Compare commits

...
2 Commits
Author SHA1 Message Date
GeorgeandClaude Sonnet 5.5 9e85c4cee8 Logs en hora local, errores de red legibles y guía NSSM en el README
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-20 09:45:34 -05:00
GeorgeandClaude Sonnet 5.5 ec312d9461 Reiniciar U_Retry al reactivar, subscription:delete y fix de grupo huérfano
- Al reactivar una suscripción (resume o detección X -> A en el dispatcher)
  se reinicia U_Retry de sus entregas pendientes, para que no vuelva a 'X'
  en el siguiente tick.
- Nuevo comando subscription:delete: borra filtros, entregas fallidas y
  eventos de @OEVN que quedan huérfanos.
- Evita TypeError en sendPendingDeliveries cuando la suscripción pasa a 'X'
  después de armar su grupo de envío.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-02 14:46:56 -05:00
6 changed files with 207 additions and 15 deletions

No files matched your search

+1
View File
@@ -3,3 +3,4 @@ node_modules
.env
logs/
tmp-test-events/
.env.test
+62 -7
View File
@@ -100,15 +100,70 @@ npm run run-once # una sola pasada, útil para probar
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
Cada tick escribe a `logs/dispatcher-YYYY-MM-DD.log` (uno por día) además
de la consola — línea simple `<timestamp ISO> [NIVEL] mensaje`, legible
por cualquier visualizador genérico (`tail -f logs/dispatcher-*.log`,
`lnav`, etc.). 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`).
Cada tick escribe a `logs/dispatcher-YYYY-MM-DD.log` (uno por día,
rotando a medianoche hora local del server) además de la consola — línea
simple `<timestamp ISO con offset local> [NIVEL] mensaje` (ej.
`2026-09-16T13:50:32.565-03:00 [INFO] ...`), legible por cualquier
visualizador genérico (`tail -f logs/dispatcher-*.log`, `lnav`, etc.). Se
usa hora local (no UTC) a propósito para que los logs sean auditables
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
+41 -2
View File
@@ -8,6 +8,7 @@
// 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: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:remove <webhookId> <bizObject> <transType>
@@ -15,6 +16,8 @@
import { config } from '../src/config.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 { userTableEntity } from '../src/lib/service-layer-paths.ts';
@@ -87,7 +90,43 @@ async function main() {
const [webhookId] = args;
if (!webhookId) fail('subscription:resume <webhookId>');
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;
}
@@ -140,7 +179,7 @@ async function main() {
default:
fail(
`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',
);
}
}
+21
View File
@@ -70,3 +70,24 @@ export async function markFailedAttempt(client: ServiceLayerClient, row: Evn1Row
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}')`);
}
+18 -1
View File
@@ -59,6 +59,11 @@ async function registerNewEvents(client: ServiceLayerClient): Promise<number> {
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>();
async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number> {
const pending = await evn1.listPending(client);
if (pending.length === 0) return 0;
@@ -66,11 +71,20 @@ 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;
@@ -83,7 +97,10 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise<number
}
for (const [webhookId, rows] of byWebhookId) {
await sendGroup(client, activeSubscriptions.get(webhookId)!, rows);
// 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);
}
await recomputeEventStatuses(client, [...new Set(pending.map((r) => r.eventId))]);
+64 -5
View File
@@ -1,5 +1,6 @@
import { appendFileSync, mkdirSync } from 'node:fs';
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
// consola — para que un servicio Windows (NSSM/PM2/Task Scheduler, ver
@@ -9,14 +10,65 @@ import { fileURLToPath } from 'node:url';
const LOGS_DIR = fileURLToPath(new URL('../../logs/', import.meta.url));
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 {
const day = date.toISOString().slice(0, 10);
const day = localTimestamp(date).slice(0, 10);
return `${LOGS_DIR}dispatcher-${day}.log`;
}
function write(level: string, message: string): void {
const now = new Date();
appendFileSync(logFilePath(now), `${now.toISOString()} [${level}] ${message}\n`);
appendFileSync(logFilePath(now), `${localTimestamp(now)} [${level}] ${message}\n`);
}
// 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 = {
@@ -29,8 +81,15 @@ export const logger = {
write('WARN', message);
},
error(message: string, err?: unknown): void {
const detail = err instanceof Error ? (err.stack ?? err.message) : err !== undefined ? String(err) : '';
console.error(message, err ?? '');
write('ERROR', detail ? `${message} ${detail}` : message);
const summary = describeError(err);
const headline = summary ? `${message} ${summary}` : 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}`);
},
};