From 97f3991cb02b39e2fbcc250f97c31e50e371f608 Mon Sep 17 00:00:00 2001 From: Horux Dev Date: Wed, 26 Aug 2026 16:45:33 +0000 Subject: [PATCH] feat(sat-sync): excluir tenant DEMO2501019X2 de sincronizaciones SAT MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Agrega SAT_SYNC_EXCLUDED_TENANTS en ecosystem.config.js con DEMO2501019X2. - Filtra tenants excluidos en getTenantsWithFiel() para crons automáticos. - Rechaza startSync() manual para tenants excluidos. - Incluye script check-daily-sync.js para monitorear cobertura diaria. --- apps/api/scripts/check-daily-sync.js | 143 ++++++++++++ apps/api/src/jobs/sat-sync.job.ts | 73 ++++--- apps/api/src/services/sat/sat.service.ts | 263 ++++++++++++++++------- ecosystem.config.js | 1 + 4 files changed, 377 insertions(+), 103 deletions(-) create mode 100644 apps/api/scripts/check-daily-sync.js diff --git a/apps/api/scripts/check-daily-sync.js b/apps/api/scripts/check-daily-sync.js new file mode 100644 index 0000000..1ef1788 --- /dev/null +++ b/apps/api/scripts/check-daily-sync.js @@ -0,0 +1,143 @@ +const { Client } = require('./node_modules/pg'); + +const HORUX_CONN = { + host: 'localhost', + port: 5432, + user: 'postgres', + password: 'ZxHMrmnwanvLfLDdNJdRthFjWF2Lj1Rb', + database: 'horux360', +}; + +const todayStart = new Date('2026-08-26T06:00:00.000Z'); // 00:00 CDMX approx +const todayEnd = new Date('2026-08-27T06:00:00.000Z'); + +async function query(client, sql, params = []) { + const res = await client.query(sql, params); + return res.rows; +} + +async function main() { + const client = new Client(HORUX_CONN); + await client.connect(); + + // Tenants activos y su FIEL legacy (tenant-level, Horux 360 antiguo) + // Se excluyen cuentas demo/ficticias configuradas en SAT_SYNC_EXCLUDED_TENANTS. + const excluded = new Set( + (process.env.SAT_SYNC_EXCLUDED_TENANTS || '') + .split(',') + .map(s => s.trim().toUpperCase()) + .filter(Boolean) + ); + + const tenants = await query(client, ` + SELECT t.id, t.rfc, t.nombre, t.database_name, + EXISTS ( + SELECT 1 FROM fiel_credentials fc + WHERE fc.tenant_id = t.id AND fc.is_active = true AND fc.valid_until > NOW() + ) AS has_legacy_fiel + FROM tenants t + WHERE t.active = true + ORDER BY t.rfc + `); + + const visibleTenants = tenants.filter(t => !excluded.has(t.id) && !excluded.has(t.rfc.toUpperCase())); + + // Daily jobs de hoy + const dailyJobs = await query(client, ` + SELECT tenant_id, contribuyente_id, cfdis_found, status, created_at + FROM sat_sync_jobs + WHERE type = 'daily' + AND created_at >= $1 + AND created_at < $2 + `, [todayStart, todayEnd]); + + const dailyByTenantContrib = new Map(); + for (const j of dailyJobs) { + const key = j.contribuyente_id ? `${j.tenant_id}:${j.contribuyente_id}` : `${j.tenant_id}:`; + dailyByTenantContrib.set(key, j); + } + + const hadDaily = []; + const missing = []; + const noFiel = []; + + for (const t of visibleTenants) { + if (!t.database_name) { + // Legacy tenant-wide (sin contribuyentes) + if (!t.has_legacy_fiel) { + noFiel.push({ tenantRfc: t.rfc, tenantName: t.nombre }); + continue; + } + const key = `${t.id}:`; + const job = dailyByTenantContrib.get(key); + if (job) { + hadDaily.push({ tenantRfc: t.rfc, tenantName: t.nombre, contribuyenteId: null, contribuyenteRfc: null, contribuyenteName: null, cfdisFound: job.cfdis_found, status: job.status }); + } else { + missing.push({ tenantRfc: t.rfc, tenantName: t.nombre, contribuyenteId: null, contribuyenteRfc: null, contribuyenteName: null, reason: 'tenant legacy con FIEL pero sin daily' }); + } + continue; + } + + // Conectar a la BD del tenant + const tenantClient = new Client({ ...HORUX_CONN, database: t.database_name }); + try { + await tenantClient.connect(); + + // Todos los contribuyentes con info de FIEL (per-contribuyente o legacy) + const contribuyentes = await query(tenantClient, ` + SELECT c.entidad_id, c.rfc, eg.nombre AS razon_social, + EXISTS ( + SELECT 1 FROM fiel_contribuyente f + WHERE f.contribuyente_id = c.entidad_id + AND f.is_active = true AND f.valid_until > NOW() + ) AS has_fiel + FROM contribuyentes c + JOIN entidades_gestionadas eg ON eg.id = c.entidad_id + ORDER BY eg.nombre + `); + + for (const c of contribuyentes) { + const eligible = c.has_fiel || t.has_legacy_fiel; + const key = `${t.id}:${c.entidad_id}`; + const job = dailyByTenantContrib.get(key); + + if (job) { + hadDaily.push({ tenantRfc: t.rfc, tenantName: t.nombre, contribuyenteId: c.entidad_id, contribuyenteRfc: c.rfc, contribuyenteName: c.razon_social, cfdisFound: job.cfdis_found, status: job.status }); + } else if (eligible) { + missing.push({ tenantRfc: t.rfc, tenantName: t.nombre, contribuyenteId: c.entidad_id, contribuyenteRfc: c.rfc, contribuyenteName: c.razon_social, reason: 'sin daily' }); + } else { + noFiel.push({ tenantRfc: t.rfc, tenantName: t.nombre, contribuyenteId: c.entidad_id, contribuyenteRfc: c.rfc, contribuyenteName: c.razon_social }); + } + } + } catch (err) { + missing.push({ tenantRfc: t.rfc, tenantName: t.nombre, contribuyenteId: null, contribuyenteRfc: null, contribuyenteName: null, reason: `error consultando BD: ${err.message}` }); + } finally { + try { await tenantClient.end(); } catch {} + } + } + + await client.end(); + + console.log('=== CONTRIBUYENTES/TENANTS CON DAILY HOY ==='); + console.log(`Total con daily: ${hadDaily.length}`); + for (const h of hadDaily) { + const name = h.contribuyenteName ? `${h.contribuyenteName} (${h.contribuyenteRfc})` : h.tenantName; + console.log(`- ${h.tenantRfc}: ${name} | CFDIs: ${h.cfdisFound ?? 'N/A'} | status: ${h.status}`); + } + + console.log('\n=== CONTRIBUYENTES/TENANTS SIN DAILY HOY (con FIEL) ==='); + console.log(`Total sin daily: ${missing.length}`); + for (const m of missing) { + const name = m.contribuyenteName ? `${m.contribuyenteName} (${m.contribuyenteRfc})` : m.tenantName; + console.log(`- ${m.tenantRfc}: ${name} [${m.reason}]`); + } + + console.log('\n=== CONTRIBUYENTES/TENANTS SIN FIEL CONFIGURADA ==='); + console.log(`Total sin FIEL: ${noFiel.length}`); + for (const n of noFiel) { + const name = n.contribuyenteName ? `${n.contribuyenteName} (${n.contribuyenteRfc})` : n.tenantName; + console.log(`- ${n.tenantRfc}: ${name}`); + } +} + +main().catch(err => { console.error(err); process.exit(1); }); diff --git a/apps/api/src/jobs/sat-sync.job.ts b/apps/api/src/jobs/sat-sync.job.ts index f3e9dd2..be0a111 100644 --- a/apps/api/src/jobs/sat-sync.job.ts +++ b/apps/api/src/jobs/sat-sync.job.ts @@ -11,18 +11,40 @@ import { consultarConstancia, purgeConstanciasAntiguas } from '../services/const import { tenantDb } from '../config/database.js'; import type { Pool } from 'pg'; -const SYNC_CRON_SCHEDULE = '0 6-10 * * *'; // 6:00–10:00 AM CDMX — ~20% de tenants por hora (5 grupos); el SAT cierra el servicio en la noche -const RECOVERY_CRON_SCHEDULE = '0 10 * * *'; // 10:00 AM todos los días -const RETRY_9AM_CRON_SCHEDULE = '0 9 * * *'; // 9:00 AM todos los días +const SYNC_CRON_SCHEDULE = '0 3-7 * * *'; // 3:00–7:00 AM CDMX — ~20% de tenants por hora (5 grupos); ventana de menor carga +const RECOVERY_CRON_SCHEDULE = '0 14 * * *'; // 2:00 PM todos los días const RETRY_4PM_CRON_SCHEDULE = '0 16 * * *'; // 4:00 PM todos los días const CONCURRENT_SYNCS = 3; // Máximo de sincronizaciones simultáneas (legacy, se mantiene por compatibilidad) -const CONCURRENT_CONTRIBUYENTES = Number(process.env.SAT_CONCURRENT_CONTRIBUYENTES || '10'); // Máximo de contribuyentes en paralelo +const CONCURRENT_CONTRIBUYENTES = Number(process.env.SAT_CONCURRENT_CONTRIBUYENTES || '5'); // Máximo de contribuyentes en paralelo const OPINION_CRON_SCHEDULE = '0 4 * * 0'; // Sundays 4:00 AM const CSF_CRON_SCHEDULE = '0 4 1 * *'; // Día 1 de cada mes 04:00 AM (CSF mensual) -const INCREMENTAL_CRON_SCHEDULE = '0 11,15,19 * * *'; // 11:00, 15:00 y 19:00; fuera de ese rango el daily (6-10 AM) cubre +const INCREMENTAL_CRON_SCHEDULE = '0 12,16 * * *'; // 12:00 PM y 4:00 PM CDMX +const BATCH_JITTER_MS = Number(process.env.SAT_BATCH_JITTER_MS || '60000'); // Jitter máximo entre inicios de sync en un batch (default 60s) const SUBSCRIPTION_LIFECYCLE_CRON = '30 2 * * *'; // 2:30 AM diario — aplica pending changes + expira trials const EXPIRY_REMINDERS_CRON = '0 9 * * *'; // 9:00 AM diario — avisos pre-vencimiento (7d/3d/1d/0d) +// Tenants excluidos de sincronizaciones SAT (cuentas demo/ficticias). +// Ejemplo: SAT_SYNC_EXCLUDED_TENANTS=DEMO2501019X2,OTRO_RFC +const EXCLUDED_TENANTS = new Set( + (process.env.SAT_SYNC_EXCLUDED_TENANTS || '') + .split(',') + .map(s => s.trim().toUpperCase()) + .filter(Boolean) +); + +function isTenantExcluded(tenantId: string, tenantRfc?: string): boolean { + return EXCLUDED_TENANTS.has(tenantId) || (tenantRfc ? EXCLUDED_TENANTS.has(tenantRfc.toUpperCase()) : false); +} + +/** + * Pausa aleatoria para desalinear el inicio de syncs dentro de un batch. + * El jitter se aplica antes de cada syncUnit para evitar ráfagas. + */ +function jitterSleep(): Promise { + const ms = Math.floor(Math.random() * BATCH_JITTER_MS); + return new Promise(resolve => setTimeout(resolve, ms)); +} + let isRunning = false; let isIncrementalRunning = false; let isRecoveryRunning = false; @@ -65,12 +87,16 @@ async function hasAnyFielConfigured(tenantId: string, databaseName?: string | nu async function getTenantsWithFiel(): Promise { const tenants = await prisma.tenant.findMany({ where: { active: true }, - select: { id: true, databaseName: true }, + select: { id: true, rfc: true, databaseName: true }, }); const tenantsWithFiel: string[] = []; for (const tenant of tenants) { + if (isTenantExcluded(tenant.id, tenant.rfc)) { + console.log(`[SAT Cron] Tenant ${tenant.rfc || tenant.id} excluido de sincronizaciones SAT`); + continue; + } const hasFiel = await hasAnyFielConfigured(tenant.id, tenant.databaseName); if (hasFiel) { tenantsWithFiel.push(tenant.id); @@ -288,8 +314,8 @@ async function syncTenant(tenantId: string): Promise { /** * Ejecuta el job de sincronización para todos los tenants */ -const DAILY_GROUPS = 5; // ventanas 6,7,8,9,10 AM -const DAILY_WINDOW_START = 6; // primera ventana CDMX +const DAILY_GROUPS = 5; // ventanas 3,4,5,6,7 AM +const DAILY_WINDOW_START = 3; // primera ventana CDMX /** Hash estable del tenantId → grupo 0..DAILY_GROUPS-1 (reparte ~20% por ventana) */ function tenantGroup(tenantId: string): number { @@ -328,7 +354,7 @@ async function runSyncJob(): Promise { } const hour = cdmxHour(); - const groupIndex = hour - DAILY_WINDOW_START; // 6→0 … 10→4 + const groupIndex = hour - DAILY_WINDOW_START; // 3→0 … 7→4 if (groupIndex < 0 || groupIndex >= DAILY_GROUPS) { console.log(`[SAT Cron] Hora CDMX ${hour} fuera de ventana 6-10 AM, omitiendo`); return; @@ -353,7 +379,10 @@ async function runSyncJob(): Promise { // Procesar en lotes de contribuyentes para aprovechar los proxies for (let i = 0; i < units.length; i += CONCURRENT_CONTRIBUYENTES) { const batch = units.slice(i, i + CONCURRENT_CONTRIBUYENTES); - await Promise.all(batch.map(unit => syncUnit(unit, '[SAT Cron]'))); + await Promise.all(batch.map(async (unit) => { + await jitterSleep(); // desalinear inicios y reducir ráfagas + return syncUnit(unit, '[SAT Cron]'); + })); // Pequeña pausa entre lotes if (i + CONCURRENT_CONTRIBUYENTES < units.length) { @@ -499,7 +528,10 @@ async function runIncrementalSyncJob(): Promise { for (let i = 0; i < units.length; i += CONCURRENT_CONTRIBUYENTES) { const batch = units.slice(i, i + CONCURRENT_CONTRIBUYENTES); - await Promise.all(batch.map(unit => syncUnit(unit, '[SAT Cron Inc]'))); + await Promise.all(batch.map(async (unit) => { + await jitterSleep(); // desalinear inicios y reducir ráfagas + return syncUnit(unit, '[SAT Cron Inc]'); + })); if (i + CONCURRENT_CONTRIBUYENTES < units.length) { await new Promise(resolve => setTimeout(resolve, 5000)); @@ -742,7 +774,6 @@ async function runDailyRetryJob(): Promise { let scheduledTask: ReturnType | null = null; let retryTask: ReturnType | null = null; let recoveryTask: ReturnType | null = null; -let retry9amTask: ReturnType | null = null; let retry4pmTask: ReturnType | null = null; let opinionTask: ReturnType | null = null; let csfTask: ReturnType | null = null; @@ -798,17 +829,7 @@ export function startSatSyncJob(): void { }); // Retomar jobs diarios que quedaron pending por timeout de polling. - // 9:00 AM y 4:00 PM CDMX, complemento a los retries automáticos de 6h/12h. - retry9amTask = cron.schedule(RETRY_9AM_CRON_SCHEDULE, async () => { - try { - await runDailyRetryJob(); - } catch (error: any) { - console.error('[SAT Daily Retry 9AM] Error:', error.message); - } - }, { - timezone: 'America/Mexico_City', - }); - + // 4:00 PM CDMX, complemento a los retries automáticos de 6h/12h. retry4pmTask = cron.schedule(RETRY_4PM_CRON_SCHEDULE, async () => { try { await runDailyRetryJob(); @@ -925,7 +946,7 @@ export function startSatSyncJob(): void { console.log(`[SAT Cron] Job programado para: ${SYNC_CRON_SCHEDULE} (America/Mexico_City)`); console.log(`[SAT Cron] Retry programado cada hora`); console.log(`[SAT Recovery Cron] Programado para: ${RECOVERY_CRON_SCHEDULE} (America/Mexico_City)`); - console.log(`[SAT Daily Retry] Programado para: ${RETRY_9AM_CRON_SCHEDULE} y ${RETRY_4PM_CRON_SCHEDULE} (America/Mexico_City)`); + console.log(`[SAT Daily Retry] Programado para: ${RETRY_4PM_CRON_SCHEDULE} (America/Mexico_City)`); console.log(`[Opinion Cron] Programado para: ${OPINION_CRON_SCHEDULE} (America/Mexico_City)`); console.log(`[CSF Cron] Programado para: ${CSF_CRON_SCHEDULE} (America/Mexico_City)`); console.log(`[SAT Cron Inc] Incremental Enterprise programado para: ${INCREMENTAL_CRON_SCHEDULE} (America/Mexico_City)`); @@ -949,10 +970,6 @@ export function stopSatSyncJob(): void { recoveryTask.stop(); recoveryTask = null; } - if (retry9amTask) { - retry9amTask.stop(); - retry9amTask = null; - } if (retry4pmTask) { retry4pmTask.stop(); retry4pmTask = null; diff --git a/apps/api/src/services/sat/sat.service.ts b/apps/api/src/services/sat/sat.service.ts index 734d295..f71a179 100644 --- a/apps/api/src/services/sat/sat.service.ts +++ b/apps/api/src/services/sat/sat.service.ts @@ -7,10 +7,13 @@ import { querySat, verifySatRequest, downloadSatPackage, + startOfMexicoDay, + SatNetworkError, type FielData, type ProxyInfo, } from './sat-client.service.js'; import { processPackage, processMetadataPackage, extractXmlsFromZip, type CfdiParsed, type CfdiMetadata } from './sat-parser.service.js'; +import { proxyManager } from './proxy.service.js'; import { recomputarSaldoPendiente, uuidsAfectadosPorCfdi } from '../../utils/saldo.js'; import type { SatSyncJob, CfdiSyncType, SatSyncType } from '@horux/shared'; import type { Service } from '@nodecfdi/sat-ws-descarga-masiva'; @@ -18,10 +21,54 @@ import type { Pool } from 'pg'; import * as fs from 'fs'; import * as path from 'path'; -const POLL_INTERVAL_MS = 5 * 60 * 1000; // 5 minutos entre verificaciones -const MAX_POLL_ATTEMPTS = 9; // 9 intentos máximo por solicitud (~45 min total) -const DAILY_MAX_POLL_ATTEMPTS = 9; // igual para daily: 9 intentos × 5 min +/** + * Intervalos de polling con backoff corto (en minutos). + * Patrón: 5, 5, 5, 10, 10, 15, 15, 20, 20 minutos. + * Reduce la presión inicial sobre el SAT y sigue dando hasta ~105 min de espera. + */ +const POLL_INTERVALS_MIN = [5, 5, 5, 10, 10, 15, 15, 20, 20]; +const POLL_INTERVALS_MS = POLL_INTERVALS_MIN.map(m => m * 60 * 1000); +const MAX_POLL_ATTEMPTS = POLL_INTERVALS_MS.length; // 9 intentos máximo por solicitud +const DAILY_MAX_POLL_ATTEMPTS = MAX_POLL_ATTEMPTS; // igual para daily const YEARS_TO_SYNC = 6; // SAT solo permite descargar últimos 6 años +const SAT_MAX_RANGE_MONTHS = 71; // 5 años y 11 meses: margen seguro bajo el límite de 6 años del SAT + +// Tenants excluidos de sincronizaciones SAT (cuentas demo/ficticias). +// Ejemplo: SAT_SYNC_EXCLUDED_TENANTS=DEMO2501019X2,OTRO_RFC +const EXCLUDED_TENANTS = new Set( + (process.env.SAT_SYNC_EXCLUDED_TENANTS || '') + .split(',') + .map(s => s.trim().toUpperCase()) + .filter(Boolean) +); + +async function isTenantExcludedFromSync(tenantId: string): Promise { + if (EXCLUDED_TENANTS.has(tenantId)) return true; + const tenant = await prisma.tenant.findUnique({ where: { id: tenantId }, select: { rfc: true } }); + return tenant?.rfc ? EXCLUDED_TENANTS.has(tenant.rfc.toUpperCase()) : false; +} + +/** + * Recorta una fecha de inicio personalizada para que el rango [dateFrom, fechaFin] + * no exceda ~5 años 11 meses. El SAT rechaza solicitudes "mayor a 6 años de + * antigüedad"; este margen evita el error 301 en los chunks más antiguos. + */ +function clampDateFrom(fechaFin: Date, customDateFrom?: Date): Date { + const limite = new Date( + fechaFin.getFullYear(), + fechaFin.getMonth() - SAT_MAX_RANGE_MONTHS, + fechaFin.getDate(), + fechaFin.getHours(), + fechaFin.getMinutes(), + fechaFin.getSeconds(), + fechaFin.getMilliseconds(), + ); + if (!customDateFrom || customDateFrom < limite) { + console.log(`[SAT] Rango recortado a ${limite.toISOString()} para no exceder 5 años 11 meses`); + return limite; + } + return customDateFrom; +} /** * Fecha final segura para consultas al SAT. @@ -100,6 +147,40 @@ interface SyncContext { getPool: () => Promise; } +/** + * Ejecuta una operación SAT reintentando con otro proxy si falla por error de red + * (timeout, ECONNRESET, 502, etc.). Actualiza el Service y proxyInfo del contexto + * para que las siguientes operaciones del mismo job usen el proxy que funcionó. + */ +async function withSatRetry( + ctx: SyncContext, + operation: (service: Service) => Promise, + maxProxyRetries = 2, +): Promise { + let lastError: any; + for (let attempt = 0; attempt <= maxProxyRetries; attempt++) { + try { + return await operation(ctx.service); + } catch (error: any) { + lastError = error; + if (!(error instanceof SatNetworkError)) { + throw error; + } + const current = ctx.proxyInfo; + const fallback = proxyManager.getDifferentProxy(current); + if (!fallback) { + console.warn(`[SAT] No hay proxy alternativo disponible para failover`); + throw error; + } + console.warn(`[SAT] Proxy failover intento ${attempt + 1}: ${current?.host}:${current?.port} → ${fallback.host}:${fallback.port}`); + const { service, proxyInfo } = createSatService(ctx.fielData, fallback); + ctx.service = service; + ctx.proxyInfo = proxyInfo; + } + } + throw lastError; +} + /** * Registra un error de bloqueo/devolución del SAT junto con el proxy usado. * Permite reportes diarios de errores por IP. @@ -660,48 +741,50 @@ async function requestAndDownload( }); let existingMap = (jobRow?.satRequestIds as Record | null) || {}; - // NOTA: se desactivó la reutilización de requestIds de jobs previos porque el SAT - // limita las descargas por solicitud. Reusar un requestId de un job anterior puede - // agotar el límite y devolver "Máximo de descargas permitidas", dejando el recovery - // sin poder descargar. Cada job nuevo crea sus propias solicitudes. - // - // Si no existe en el job actual, buscar en el job más reciente del mismo tenant/contribuyente - // SOLO si el rango de fechas es idéntico (mismo dateFrom/dateTo). - // if (!existingMap[kindKey]) { - // const previousJob = await prisma.satSyncJob.findFirst({ - // where: { - // tenantId: jobRow?.tenantId, - // contribuyenteId: jobRow?.contribuyenteId ?? null, - // id: { not: jobId }, - // dateFrom: jobRow?.dateFrom, - // dateTo: jobRow?.dateTo, - // }, - // orderBy: { createdAt: 'desc' }, - // select: { satRequestIds: true }, - // }); - // if (previousJob?.satRequestIds) { - // const prevMap = previousJob.satRequestIds as Record; - // if (prevMap[kindKey]) { - // console.log(`[SAT] Reutilizando requestId de job previo (${label}): ${prevMap[kindKey]}`); - // // Copiar al job actual para futuros usos - // await persistSatRequestId(jobId, kindKey, prevMap[kindKey]); - // existingMap = { ...existingMap, [kindKey]: prevMap[kindKey] }; - // } - // } - // } + // Reutilización segura de requestIds entre jobs recientes: + // - Solo jobs del mismo tenant/contribuyente, tipo daily/incremental, estado completed. + // - Solo si el requestId fue creado en las últimas 48h (evita expiración >72h del SAT). + // - Solo si el kindKey (rango+tipo+requestType) coincide exactamente. + // Esto ahorra quota del SAT y evita crear solicitudes duplicadas para el mismo rango. + const REQUEST_ID_REUSE_MAX_AGE_HOURS = 48; + if (!existingMap[kindKey]) { + const reuseCutoff = new Date(Date.now() - REQUEST_ID_REUSE_MAX_AGE_HOURS * 60 * 60 * 1000); + const previousJob = await prisma.satSyncJob.findFirst({ + where: { + tenantId: jobRow?.tenantId, + contribuyenteId: jobRow?.contribuyenteId ?? null, + id: { not: jobId }, + type: { in: ['daily', 'incremental'] }, + status: 'completed', + createdAt: { gte: reuseCutoff }, + }, + orderBy: { createdAt: 'desc' }, + select: { satRequestIds: true }, + }); + if (previousJob?.satRequestIds) { + const prevMap = previousJob.satRequestIds as Record; + if (prevMap[kindKey]) { + console.log(`[SAT] Reutilizando requestId de job previo (${label}): ${prevMap[kindKey]}`); + await persistSatRequestId(jobId, kindKey, prevMap[kindKey]); + existingMap = { ...existingMap, [kindKey]: prevMap[kindKey] }; + } + } + } let requestId: string | null = existingMap[kindKey] || null; let verifyResult: Awaited> | undefined; if (requestId) { - console.log(`[SAT] Reusando requestId previo (${label}): ${requestId}`); + const currentRequestId = requestId; // alias para estrechamiento de tipo + console.log(`[SAT] Reusando requestId previo (${label}): ${currentRequestId}`); try { - verifyResult = await verifySatRequest(ctx.service, requestId); + verifyResult = await withSatRetry(ctx, service => verifySatRequest(service, currentRequestId)); console.log(`[SAT] Estado del request reusado (${label}): ${verifyResult.status}`); // Estados terminales inválidos → descartar y crear nuevo if (verifyResult.status === 'failed' || verifyResult.status === 'rejected') { - if (isAgotadas(verifyResult.message)) { + if (isAgotadas(verifyResult.message) || verifyResult.codeRequest === '5002') { + await recordSatProxyError(jobId, ctx, verifyResult.codeRequest || '5002', stageIdForTimeout(label), `${label}: solicitud reusada agotada de por vida`); console.log(`[SAT] Solicitud reusada agotada de por vida (${label}); se cancela y se omite, no se recrea.`); return { packageContents: [], totalCfdis: 0 }; } @@ -722,10 +805,12 @@ async function requestAndDownload( if (!requestId) { console.log(`[SAT] Solicitando ${label} desde ${fechaInicio.toISOString()} hasta ${fechaFin.toISOString()}`); - const queryResult = await querySat(ctx.service, fechaInicio, fechaFin, tipoCfdi, requestType); + const queryResult = await withSatRetry(ctx, service => querySat(service, fechaInicio, fechaFin, tipoCfdi, requestType)); if (!queryResult.success) { - if (queryResult.statusCode === '5004' || queryResult.statusCode === 'exhausted' || isAgotadas(queryResult.message)) { + if (queryResult.statusCode === '5004' || queryResult.statusCode === '5002' || isAgotadas(queryResult.message)) { + const satCode = queryResult.statusCode === '5002' || isAgotadas(queryResult.message) ? '5002' : '5004'; + await recordSatProxyError(jobId, ctx, satCode, stageIdForTimeout(label), `${label}: ${queryResult.message}`); console.log(`[SAT] Sin CFDIs, quota agotada o solicitudes agotadas (${label}): ${queryResult.message}`); return { packageContents: [], totalCfdis: 0 }; } @@ -755,15 +840,17 @@ async function requestAndDownload( const maxAttempts = isDaily ? DAILY_MAX_POLL_ATTEMPTS : MAX_POLL_ATTEMPTS; let attempts = 0; while (attempts < maxAttempts) { - await new Promise(resolve => setTimeout(resolve, POLL_INTERVAL_MS)); + const delayMs = POLL_INTERVALS_MS[attempts]; + await new Promise(resolve => setTimeout(resolve, delayMs)); attempts++; - verifyResult = await verifySatRequest(ctx.service, requestId); - console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts}/${maxAttempts})`); + verifyResult = await withSatRetry(ctx, service => verifySatRequest(service, requestId)); + console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts}/${maxAttempts}, espera ${delayMs / 60000}min)`); if (verifyResult.status === 'ready') break; if (verifyResult.status === 'failed' || verifyResult.status === 'rejected') { - if (isAgotadas(verifyResult.message)) { + if (isAgotadas(verifyResult.message) || verifyResult.codeRequest === '5002') { + await recordSatProxyError(jobId, ctx, verifyResult.codeRequest || '5002', stageIdForTimeout(label), `${label}: solicitudes agotadas de por vida`); console.log(`[SAT] Solicitudes agotadas de por vida (${label}); se cancela y se omite este rango.`); return { packageContents: [], totalCfdis: 0 }; } @@ -777,12 +864,18 @@ async function requestAndDownload( throw new SatSyncTimeoutError(stageIdForTimeout(label), `Timeout esperando respuesta del SAT (${label})`); } + // Registrar EmptyResult (5004) / Exhausted (5002) cuando el verify termina en + // ready con 0 paquetes. Esto permite distinguir "no hay CFDIs" de quota agotada. + if (verifyResult.totalCfdis === 0 && verifyResult.packageIds.length === 0 && verifyResult.codeRequest) { + await recordSatProxyError(jobId, ctx, verifyResult.codeRequest, stageIdForTimeout(label), `${label}: ${verifyResult.message}`); + } + const packageContents: string[] = []; for (let i = 0; i < verifyResult.packageIds.length; i++) { const packageId = verifyResult.packageIds[i]; console.log(`[SAT] Descargando paquete ${label} ${i + 1}/${verifyResult.packageIds.length}: ${packageId}`); - const downloadResult = await downloadSatPackage(ctx.service, packageId); + const downloadResult = await withSatRetry(ctx, service => downloadSatPackage(service, packageId)); if (!downloadResult.success) { console.error(`[SAT] Error descargando paquete ${packageId}: ${downloadResult.message}`); continue; @@ -992,9 +1085,9 @@ async function processInitialSync( // Si el usuario pasó un rango explícito lo respetamos (aunque podría fallar // si pone "hoy"). const fechaFin = customDateTo || getYesterdayEnd(); - // Exactamente 6 años atrás desde la fecha final (mismo día del mes), no inicio de mes. - // El SAT rechaza "mayor a 6 años" si usamos el día 1 del mes hace 6 años. - const inicioHistorico = customDateFrom || new Date(fechaFin.getFullYear() - YEARS_TO_SYNC, fechaFin.getMonth(), fechaFin.getDate()); + // Inicio histórico: 5 años y 11 meses atrás. El SAT rechaza rangos "mayor a 6 años", + // así que dejamos 1 mes de margen de seguridad respecto al límite legal. + const inicioHistorico = clampDateFrom(fechaFin, customDateFrom); // Paso 1: Sondeo — determinar tamaño de bloque para XMLs const chunkMonths = await determineChunkMonths(ctx, jobId, inicioHistorico, fechaFin); @@ -1025,6 +1118,7 @@ async function processInitialSync( } // Paso 2: Descargar XMLs de vigentes (bloques de 3/6 meses) + const xmlErrors: { stage: string; bloque: number; start: string; end: string; message: string }[] = []; for (let i = 0; i < xmlChunks.length; i++) { const { start, end } = xmlChunks[i]; console.log(`[SAT] XML bloque ${i + 1}/${xmlChunks.length}: ${start.toISOString().slice(0, 10)} → ${end.toISOString().slice(0, 10)}`); @@ -1037,6 +1131,7 @@ async function processInitialSync( totalUpdated += emitidos.updated; } catch (error: any) { console.error(`[SAT] Error emitidos XML bloque ${i + 1}:`, error.message); + xmlErrors.push({ stage: 'xml-emitidos', bloque: i + 1, start: start.toISOString(), end: end.toISOString(), message: error.message }); } await reportProgress(); @@ -1048,6 +1143,7 @@ async function processInitialSync( totalUpdated += recibidos.updated; } catch (error: any) { console.error(`[SAT] Error recibidos XML bloque ${i + 1}:`, error.message); + xmlErrors.push({ stage: 'xml-recibidos', bloque: i + 1, start: start.toISOString(), end: end.toISOString(), message: error.message }); } await reportProgress(); @@ -1055,6 +1151,7 @@ async function processInitialSync( } // Paso 3: Descargar metadata (bloques anuales) + const metadataErrors: { stage: string; bloque: number; start: string; end: string; message: string }[] = []; for (let i = 0; i < metaChunks.length; i++) { const { start, end } = metaChunks[i]; console.log(`[SAT] Metadata bloque ${i + 1}/${metaChunks.length}: ${start.toISOString().slice(0, 10)} → ${end.toISOString().slice(0, 10)}`); @@ -1065,6 +1162,7 @@ async function processInitialSync( totalUpdated += updated; } catch (error: any) { console.error(`[SAT] Error metadata emitidos bloque ${i + 1}:`, error.message); + metadataErrors.push({ stage: 'metadata-emitidos', bloque: i + 1, start: start.toISOString(), end: end.toISOString(), message: error.message }); } await reportProgress(); @@ -1074,6 +1172,7 @@ async function processInitialSync( totalUpdated += updated; } catch (error: any) { console.error(`[SAT] Error metadata recibidos bloque ${i + 1}:`, error.message); + metadataErrors.push({ stage: 'metadata-recibidos', bloque: i + 1, start: start.toISOString(), end: end.toISOString(), message: error.message }); } await reportProgress(); @@ -1086,7 +1185,16 @@ async function processInitialSync( cfdisInserted: totalInserted, cfdisUpdated: totalUpdated, progressPercent: 100, + errorMessage: xmlErrors.length > 0 || metadataErrors.length > 0 + ? JSON.stringify({ xmlErrors, metadataErrors }, null, 2) + : undefined, }); + + // Si fallo alguna etapa de XMLs, marcamos el job como failed para que sea + // facil identificarlo y repetir la sincronizacion mas tarde. + if (xmlErrors.length > 0) { + throw new Error(`Fallo en etapa de XMLs: ${xmlErrors.map(e => `${e.stage} bloque ${e.bloque}`).join(', ')}`); + } } /** @@ -1100,9 +1208,10 @@ async function processInitialSync( async function processCustomRangeSync( ctx: SyncContext, jobId: string, - dateFrom: Date, + dateFromInput: Date, dateTo: Date ): Promise { + const dateFrom = clampDateFrom(dateTo, dateFromInput); const diffMs = dateTo.getTime() - dateFrom.getTime(); const diffMonths = diffMs / (1000 * 60 * 60 * 24 * 30); const MAX_MONTHS_DIRECT = 6; @@ -1175,41 +1284,37 @@ async function processCustomRangeSync( } /** - * Sincronización incremental: ventana fija de las últimas 8 horas. - * Diseñada para correr 3 veces al día (11:00, 15:00, 19:00) en clientes Enterprise. - * La ventana de 8h cubre el gap máximo (03:00 → 11:00) entre el daily y el primer - * incremental; los disparos siguientes solapan, pero la unicidad del UUID deduplica. - * Fuera de ese rango, el daily de 03:00 se encarga. + * Sincronización incremental: rango ayer → hoy (hora CDMX). + * Corre 3 veces al día (11:00, 15:00 y 19:00) en clientes Enterprise. + * El SAT solo acepta rangos de días calendario distintos, así que la ventana + * mínima válida es ayer 00:00 → hoy 00:00. Cada ejecución recupera las + * facturas de hoy que el SAT ya haya procesado desde la última corrida. + * + * Solo descargamos XMLs (cfdi); la metadata histórica la cubre el daily + * (domingos) y el recovery cron. Incluir metadata en incremental hace que + * el job exceda el umbral del watchdog porque los paquetes de metadata de + * 1 día tardan demasiado en generarse. + * Los disparos solapan, pero la unicidad del UUID deduplica. */ -const INCREMENTAL_WINDOW_HOURS = 8; - async function processIncrementalSync(ctx: SyncContext, jobId: string): Promise { - // Retrocedemos 2h respecto a ahora para evitar que el SAT vea una fecha final - // futura / demasiado reciente (rechazo "Fecha final invalida"). - const ahora = new Date(Date.now() - 2 * 60 * 60 * 1000); - const desde = new Date(ahora.getTime() - INCREMENTAL_WINDOW_HOURS * 60 * 60 * 1000); + const fechaFin = startOfMexicoDay(0); // hoy 00:00 CDMX + const desde = startOfMexicoDay(-1); // ayer 00:00 CDMX let totalFound = 0; let totalDownloaded = 0; let totalInserted = 0; let totalUpdated = 0; - console.log(`[SAT] Incremental: ${desde.toISOString()} → ${ahora.toISOString()} (${INCREMENTAL_WINDOW_HOURS}h)`); + console.log(`[SAT] Incremental: ${desde.toISOString()} → ${fechaFin.toISOString()} (ayer → hoy CDMX)`); for (const tipo of ['emitidos', 'recibidos'] as const) { - const result = await processDateRange(ctx, jobId, desde, ahora, tipo); + const result = await processDateRange(ctx, jobId, desde, fechaFin, tipo); totalFound += result.found; totalDownloaded += result.downloaded; totalInserted += result.inserted; totalUpdated += result.updated; } - for (const tipo of ['emitidos', 'recibidos'] as const) { - const { inserted, updated } = await processMetadataRange(ctx, jobId, desde, ahora, tipo); - totalInserted += inserted; - totalUpdated += updated; - } - await updateJobProgress(jobId, { cfdisFound: totalFound, cfdisDownloaded: totalDownloaded, @@ -1483,15 +1588,17 @@ async function requestAndDownloadWithId( let attempts = 0; const maxAttempts = isDaily ? DAILY_MAX_POLL_ATTEMPTS : MAX_POLL_ATTEMPTS; while (attempts < maxAttempts) { - await new Promise(resolve => setTimeout(resolve, POLL_INTERVAL_MS)); + const delayMs = POLL_INTERVALS_MS[attempts]; + await new Promise(resolve => setTimeout(resolve, delayMs)); attempts++; - verifyResult = await verifySatRequest(ctx.service, requestId); - console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts}/${maxAttempts})`); + verifyResult = await withSatRetry(ctx, service => verifySatRequest(service, requestId)); + console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts}/${maxAttempts}, espera ${delayMs / 60000}min)`); if (verifyResult.status === 'ready') break; if (verifyResult.status === 'failed' || verifyResult.status === 'rejected') { - if (isAgotadas(verifyResult.message)) { + if (isAgotadas(verifyResult.message) || verifyResult.codeRequest === '5002') { + await recordSatProxyError(jobId, ctx, verifyResult.codeRequest || '5002', stageIdForTimeout(label), `${label}: solicitudes agotadas de por vida`); console.log(`[SAT] Solicitudes agotadas de por vida (${label}); se cancela y se omite.`); return { inserted: 0, updated: 0 }; } @@ -1507,7 +1614,7 @@ async function requestAndDownloadWithId( const packageId = verifyResult.packageIds[i]; console.log(`[SAT] Descargando paquete ${label} ${i + 1}/${verifyResult.packageIds.length}: ${packageId}`); - const downloadResult = await downloadSatPackage(ctx.service, packageId); + const downloadResult = await withSatRetry(ctx, service => downloadSatPackage(service, packageId)); if (!downloadResult.success) { console.error(`[SAT] Error descargando paquete ${packageId}: ${downloadResult.message}`); continue; @@ -1534,6 +1641,10 @@ export async function startSync( dateTo?: Date, contribuyenteId?: string ): Promise { + if (await isTenantExcludedFromSync(tenantId)) { + throw new Error('Este tenant está excluido de sincronizaciones SAT'); + } + // Try per-contribuyente FIEL first (despachos), then legacy (Horux360) let decryptedFiel = null; if (contribuyenteId) { @@ -1585,16 +1696,18 @@ export async function startSync( const now = new Date(); // `isCustomRange` solo aplica a 'initial' con fechas explícitas del UI. - // Bootstrap puro = initial sin fechas → usa default de 6 años atrás. + // Bootstrap puro = initial sin fechas → usa el límite seguro de 5 años 11 meses. const isCustomRange = type === 'initial' && (!!dateFrom || !!dateTo); + const effectiveDateTo = dateTo || now; + const effectiveDateFrom = clampDateFrom(effectiveDateTo, dateFrom); const job = await prisma.satSyncJob.create({ data: { tenantId, contribuyenteId: contribuyenteId || null, type, status: 'running', - dateFrom: dateFrom || new Date(now.getFullYear() - YEARS_TO_SYNC, 0, 1), - dateTo: dateTo || now, + dateFrom: effectiveDateFrom, + dateTo: effectiveDateTo, startedAt: now, isCustomRange, proxyUsed: proxyInfo ? `${proxyInfo.host}:${proxyInfo.port}` : null, @@ -1616,11 +1729,11 @@ export async function startSync( (async () => { try { if (type === 'initial') { - await processInitialSync(ctx, job.id, dateFrom, dateTo); + await processInitialSync(ctx, job.id, effectiveDateFrom, dateTo); } else if (type === 'incremental') { await processIncrementalSync(ctx, job.id); } else if (dateFrom && dateTo) { - await processCustomRangeSync(ctx, job.id, dateFrom, dateTo); + await processCustomRangeSync(ctx, job.id, effectiveDateFrom, dateTo); } else { await processDailySync(ctx, job.id); } diff --git a/ecosystem.config.js b/ecosystem.config.js index 133b73a..960f6d8 100644 --- a/ecosystem.config.js +++ b/ecosystem.config.js @@ -15,6 +15,7 @@ module.exports = { env: { NODE_ENV: 'production', PORT: 4000, + SAT_SYNC_EXCLUDED_TENANTS: 'DEMO2501019X2', }, }, {