From ec312d94618ff27fcb27a39c3d886bbc32a746df Mon Sep 17 00:00:00 2001 From: George Date: Fri, 2 Oct 2026 14:46:56 -0500 Subject: [PATCH] =?UTF-8?q?Reiniciar=20U=5FRetry=20al=20reactivar,=20subsc?= =?UTF-8?q?ription:delete=20y=20fix=20de=20grupo=20hu=C3=A9rfano?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- scripts/manage-subscriptions.ts | 43 +++++++++++++++++++++++++++++-- src/dispatcher/evn1-repository.ts | 21 +++++++++++++++ src/dispatcher/process-tick.ts | 19 +++++++++++++- 3 files changed, 80 insertions(+), 3 deletions(-) diff --git a/scripts/manage-subscriptions.ts b/scripts/manage-subscriptions.ts index b80aa2c..70a10bd 100644 --- a/scripts/manage-subscriptions.ts +++ b/scripts/manage-subscriptions.ts @@ -8,6 +8,7 @@ // npx tsx scripts/manage-subscriptions.ts subscription:update-url // npx tsx scripts/manage-subscriptions.ts subscription:pause // npx tsx scripts/manage-subscriptions.ts subscription:resume +// npx tsx scripts/manage-subscriptions.ts subscription:delete // // npx tsx scripts/manage-subscriptions.ts filter:add // npx tsx scripts/manage-subscriptions.ts filter:remove @@ -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 '); 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 '); + + // 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', ); } } diff --git a/src/dispatcher/evn1-repository.ts b/src/dispatcher/evn1-repository.ts index ab93a21..e736578 100644 --- a/src/dispatcher/evn1-repository.ts +++ b/src/dispatcher/evn1-repository.ts @@ -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 { + 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 { + 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 { + await client.delete(`/${ENTITY}('${code}')`); +} diff --git a/src/dispatcher/process-tick.ts b/src/dispatcher/process-tick.ts index a17cfed..14152b6 100644 --- a/src/dispatcher/process-tick.ts +++ b/src/dispatcher/process-tick.ts @@ -59,6 +59,11 @@ async function registerNewEvents(client: ServiceLayerClient): Promise { 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(); + async function sendPendingDeliveries(client: ServiceLayerClient): Promise { const pending = await evn1.listPending(client); if (pending.length === 0) return 0; @@ -66,11 +71,20 @@ async function sendPendingDeliveries(client: ServiceLayerClient): Promise(); + 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 r.eventId))]);