consultoria-as/HoruxDespachosNuevo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sat): metadata histórica solo domingos, 404 no fatal en daily, polling 9x5min y docs

Browse Source 
- processDailySync ejecuta metadata histórica únicamente los domingos en CDMX.
- Errores 404 en daily se registran en error_message sin abortar el job.
- Timeouts/transitorios reales siguen lanzándose para reintento.
- Polling reducido a 9 intentos máximos cada 5 minutos por solicitud.
- Actualiza docs/SAT-SYNC-IMPLEMENTATION.md con arquitectura, crons, manejo de errores y comandos.
This commit is contained in:
Horux Dev
2026-08-02 19:22:49 +00:00
parent b5701b603c
commit dfc0183c12
2 changed files with 782 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

664
apps/api/src/services/sat/sat.service.ts
View File

@@ -17,10 +17,24 @@ import type { Pool } from 'pg';
import * as fs from 'fs';
import * as path from 'path';
const POLL_INTERVAL_MS = 60000; // 60 segundos
const MAX_POLL_ATTEMPTS = 500; // ~8 horas máximo para syncs iniciales grandes
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
const YEARS_TO_SYNC = 6; // SAT solo permite descargar últimos 6 años
/**
* Fecha final segura para consultas al SAT.
*
* El SAT rechaza fechas futuras e incluso "hoy" en algunos horarios/condiciones,
* devolviendo "Fecha final invalida". Usamos el día anterior a medio día UTC,
* que al interpretarse en America/Mexico_City siempre cae en "ayer" y evita
* tanto fechas futuras como problemas de cambio de día por zona horaria.
*/
function getYesterdayEnd(): Date {
const now = new Date();
return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() - 1, 12, 0, 0));
}
/**
* Política de retry por tipo de sync.
* - `retryAtHours[i]` = horas DESDE startedAt para el retry i+1.
@@ -49,6 +63,12 @@ const RETRY_POLICIES: Record<'daily' | 'custom' | 'initial' | 'incremental', {
incremental: { maxRetries: 0, retryAtHours: [] },
};
/**
* Límite total de intentos para jobs diarios. Incluye el intento original más
* retries automáticos (6h/12h) y los retries fijos de 9 AM / 4 PM CDMX.
*/
const MAX_DAILY_RETRY_ATTEMPTS = 5;
function getRetryPolicy(job: { type: SatSyncType; isCustomRange: boolean }) {
if (job.type === 'initial' && job.isCustomRange) return RETRY_POLICIES.custom;
return RETRY_POLICIES[job.type];
@@ -598,6 +618,7 @@ async function requestAndDownload(
fechaFin: Date,
tipoCfdi: CfdiSyncType,
requestType: 'cfdi' | 'metadata',
isDaily = false,
): Promise<{ packageContents: string[]; totalCfdis: number }> {
const label = `${tipoCfdi}/${requestType}`;
const kindKey = makeRequestKindKey(fechaInicio, fechaFin, tipoCfdi, requestType);
@@ -650,6 +671,10 @@ async function requestAndDownload(
// Estados terminales inválidos → descartar y crear nuevo
if (verifyResult.status === 'failed' || verifyResult.status === 'rejected') {
if (isAgotadas(verifyResult.message)) {
console.log(`[SAT] Solicitud reusada agotada de por vida (${label}); se cancela y se omite, no se recrea.`);
return { packageContents: [], totalCfdis: 0 };
}
console.log(`[SAT] Request reusado en estado ${verifyResult.status}, creando nuevo`);
requestId = null;
verifyResult = undefined;
@@ -670,10 +695,20 @@ async function requestAndDownload(
const queryResult = await querySat(ctx.service, fechaInicio, fechaFin, tipoCfdi, requestType);
if (!queryResult.success) {
if (queryResult.statusCode === '5004') {
console.log(`[SAT] No se encontraron CFDIs (${label})`);
if (queryResult.statusCode === '5004' || queryResult.statusCode === 'exhausted' || isAgotadas(queryResult.message)) {
console.log(`[SAT] Sin CFDIs, quota agotada o solicitudes agotadas (${label}): ${queryResult.message}`);
return { packageContents: [], totalCfdis: 0 };
}
if (/error no controlado/i.test(queryResult.message || '')) {
if (isDaily) {
// En daily no detenemos el job por un 404 transitorio del SAT; se
// registra como no fatal para diagnóstico y se continúa.
console.warn(`[SAT] Rechazo 404 del SAT en daily (${label}): ${queryResult.message} — se registra y continúa`);
throw new Error(`SAT 404 en daily (${label}): ${queryResult.message}`);
}
console.warn(`[SAT] Rechazo transitorio del SAT (${label}): ${queryResult.message} — se reintentará`);
throw new SatTransientError(stageIdForTimeout(label), queryResult.message);
}
throw new Error(`Error SAT (${label}): ${queryResult.message}`);
}
@@ -685,23 +720,28 @@ async function requestAndDownload(
// Polling — si el reuse ya devolvió `ready`, salta el loop directamente.
if (!verifyResult || verifyResult.status !== 'ready') {
const maxAttempts = isDaily ? DAILY_MAX_POLL_ATTEMPTS : MAX_POLL_ATTEMPTS;
let attempts = 0;
while (attempts < MAX_POLL_ATTEMPTS) {
while (attempts < maxAttempts) {
await new Promise(resolve => setTimeout(resolve, POLL_INTERVAL_MS));
attempts++;
verifyResult = await verifySatRequest(ctx.service, requestId);
console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts})`);
console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts}/${maxAttempts})`);
if (verifyResult.status === 'ready') break;
if (verifyResult.status === 'failed' || verifyResult.status === 'rejected') {
if (isAgotadas(verifyResult.message)) {
console.log(`[SAT] Solicitudes agotadas de por vida (${label}); se cancela y se omite este rango.`);
return { packageContents: [], totalCfdis: 0 };
}
throw new Error(`Solicitud fallida (${label}): ${verifyResult.message}`);
}
}
}
if (!verifyResult || verifyResult.status !== 'ready') {
throw new Error(`Timeout esperando respuesta del SAT (${label})`);
throw new SatSyncTimeoutError(stageIdForTimeout(label), `Timeout esperando respuesta del SAT (${label})`);
}
const packageContents: string[] = [];
@@ -741,17 +781,20 @@ async function processDateRange(
fechaInicio: Date,
fechaFin: Date,
tipoCfdi: CfdiSyncType,
skipJobUpdate = false
): Promise<{ found: number; downloaded: number; inserted: number; updated: number }> {
skipJobUpdate = false,
throwOnError = false,
isDaily = false
): Promise<{ found: number; downloaded: number; inserted: number; updated: number; errors: { message: string }[] }> {
let totalFound = 0;
let totalDownloaded = 0;
let totalInserted = 0;
let totalUpdated = 0;
const errors: { message: string }[] = [];
// Solo XMLs de vigentes (datos completos)
try {
const { packageContents, totalCfdis } = await requestAndDownload(
ctx, jobId, fechaInicio, fechaFin, tipoCfdi, 'cfdi'
ctx, jobId, fechaInicio, fechaFin, tipoCfdi, 'cfdi', isDaily
);
totalFound += totalCfdis;
@@ -766,6 +809,8 @@ async function processDateRange(
}
} catch (error: any) {
console.error(`[SAT] Error en XMLs ${tipoCfdi}: ${error.message}`);
if (throwOnError) throw error;
errors.push({ message: error.message || `Error desconocido en XMLs ${tipoCfdi}` });
}
if (!skipJobUpdate) {
@@ -782,6 +827,7 @@ async function processDateRange(
downloaded: totalDownloaded,
inserted: totalInserted,
updated: totalUpdated,
errors,
};
}
@@ -794,14 +840,17 @@ async function processMetadataRange(
jobId: string,
fechaInicio: Date,
fechaFin: Date,
tipoCfdi: CfdiSyncType
): Promise<{ inserted: number; updated: number }> {
tipoCfdi: CfdiSyncType,
throwOnError = false,
isDaily = false
): Promise<{ inserted: number; updated: number; errors: { message: string }[] }> {
let totalInserted = 0;
let totalUpdated = 0;
const errors: { message: string }[] = [];
try {
const { packageContents } = await requestAndDownload(
ctx, jobId, fechaInicio, fechaFin, tipoCfdi, 'metadata'
ctx, jobId, fechaInicio, fechaFin, tipoCfdi, 'metadata', isDaily
);
for (const content of packageContents) {
@@ -814,9 +863,11 @@ async function processMetadataRange(
}
} catch (error: any) {
console.error(`[SAT] Error en metadata ${tipoCfdi}: ${error.message}`);
if (throwOnError) throw error;
errors.push({ message: error.message || `Error desconocido en metadata ${tipoCfdi}` });
}
return { inserted: totalInserted, updated: totalUpdated };
return { inserted: totalInserted, updated: totalUpdated, errors };
}
/**
@@ -904,11 +955,13 @@ async function processInitialSync(
customDateFrom?: Date,
customDateTo?: Date
): Promise<void> {
const ahora = new Date();
// Exactamente 6 años atrás desde hoy (mismo día del mes), no inicio de mes.
// El SAT rechaza fechas futuras; por defecto usamos ayer como fecha final.
// 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(ahora.getFullYear() - YEARS_TO_SYNC, ahora.getMonth(), ahora.getDate());
const fechaFin = customDateTo || ahora;
const inicioHistorico = customDateFrom || new Date(fechaFin.getFullYear() - YEARS_TO_SYNC, fechaFin.getMonth(), fechaFin.getDate());
// Paso 1: Sondeo — determinar tamaño de bloque para XMLs
const chunkMonths = await determineChunkMonths(ctx, jobId, inicioHistorico, fechaFin);
@@ -1098,7 +1151,9 @@ async function processCustomRangeSync(
const INCREMENTAL_WINDOW_HOURS = 8;
async function processIncrementalSync(ctx: SyncContext, jobId: string): Promise<void> {
const ahora = new Date();
// 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);
let totalFound = 0;
@@ -1109,25 +1164,17 @@ async function processIncrementalSync(ctx: SyncContext, jobId: string): Promise<
console.log(`[SAT] Incremental: ${desde.toISOString()} → ${ahora.toISOString()} (${INCREMENTAL_WINDOW_HOURS}h)`);
for (const tipo of ['emitidos', 'recibidos'] as const) {
try {
const result = await processDateRange(ctx, jobId, desde, ahora, tipo);
totalFound += result.found;
totalDownloaded += result.downloaded;
totalInserted += result.inserted;
totalUpdated += result.updated;
} catch (error: any) {
console.error(`[SAT] Error incremental XMLs ${tipo}:`, error.message);
}
const result = await processDateRange(ctx, jobId, desde, ahora, tipo);
totalFound += result.found;
totalDownloaded += result.downloaded;
totalInserted += result.inserted;
totalUpdated += result.updated;
}
for (const tipo of ['emitidos', 'recibidos'] as const) {
try {
const { inserted, updated } = await processMetadataRange(ctx, jobId, desde, ahora, tipo);
totalInserted += inserted;
totalUpdated += updated;
} catch (error: any) {
console.error(`[SAT] Error incremental metadata ${tipo}:`, error.message);
}
const { inserted, updated } = await processMetadataRange(ctx, jobId, desde, ahora, tipo);
totalInserted += inserted;
totalUpdated += updated;
}
await updateJobProgress(jobId, {
@@ -1138,51 +1185,310 @@ async function processIncrementalSync(ctx: SyncContext, jobId: string): Promise<
});
}
async function processDailySync(ctx: SyncContext, jobId: string): Promise<void> {
const ahora = new Date();
/**
* Error usado para señalar que una etapa del sync diario excedió el tiempo
* de espera al SAT. Lleva el identificador de la etapa para poder retomar
* desde el mismo punto en los retries programados.
*/
class SatSyncTimeoutError extends Error {
constructor(
public readonly stageId: string,
message: string
) {
super(message);
this.name = 'SatSyncTimeoutError';
}
}
/**
* Rechazo transitorio del SAT (p. ej. "Error no controlado", típico de
* throttling cuando se lanzan muchas solicitudes en ráfaga). A diferencia de
* "solicitudes agotadas", SÍ vale la pena reintentarlo: se comporta como un
* timeout — se estaciona la etapa y los retries de 9 AM / 4 PM la retoman.
*/
class SatTransientError extends Error {
constructor(
public readonly stageId: string,
message: string
) {
super(message);
this.name = 'SatTransientError';
}
}
/**
* El daily terminó sus etapas XML y parte de las de metadata, pero uno o más
* chunks de metadata aún no tienen paquetes listos. No es un fallo del SAT:
* el requestId ya quedó persistido y los retries de 9 AM / 4 PM lo retoman.
* Se trata como "pending" (no consume el job entero) en vez de abortar.
*/
class SatMetadataPendingError extends Error {
constructor(
public readonly stageId: string,
public readonly pendingStages: string[],
message: string,
) {
super(message);
this.name = 'SatMetadataPendingError';
}
}
/**
* Detecta la respuesta del SAT "Se han agotado las solicitudes de por vida":
* se agotaron las solicitudes máximas para ese rango de fecha. No es
* transitorio — no vale la pena seguir intentando ni recrear la solicitud;
* hay que cancelarla y omitir ese rango.
*/
function isAgotadas(message?: string | null): boolean {
return !!message && message.toLowerCase().includes('agotad');
}
/**
* Genera un stageId a partir de la etiqueta de una solicitud SAT.
* Usado cuando requestAndDownload detecta timeout y no conoce el stage exacto.
*/
function stageIdForTimeout(label: string): string {
// label tiene forma "tipoCfdi/requestType" (ej. "emitidos/cfdi" o "recibidos/metadata")
const [tipo, requestType] = label.split('/');
if (requestType === 'cfdi') return `xml-${tipo}-7d`;
return `metadata-${tipo}-chunk`;
}
function isSundayInCDMX(now: Date = new Date()): boolean {
return new Intl.DateTimeFormat('en-US', {
timeZone: 'America/Mexico_City',
weekday: 'short',
}).format(now) === 'Sun';
}
async function processDailySync(
ctx: SyncContext,
jobId: string,
resumeFromStage?: string
): Promise<void> {
// Usamos ayer como fecha final para evitar el rechazo "Fecha final invalida"
// del SAT cuando se consulta con la fecha actual.
const ahora = getYesterdayEnd();
const inicioAño = new Date(ahora.getFullYear(), 0, 1);
const hace7Dias = new Date(ahora.getTime() - 7 * 24 * 60 * 60 * 1000);
const ejecutarMetadataHistorica = isSundayInCDMX();
let totalFound = 0;
let totalDownloaded = 0;
let totalInserted = 0;
let totalUpdated = 0;
// Paso 1: XMLs de los últimos 7 días (CFDIs nuevos)
console.log(`[SAT] Daily: XMLs desde ${hace7Dias.toISOString().slice(0, 10)} → ${ahora.toISOString().slice(0, 10)}`);
interface DailyStage {
id: string;
label: string;
isMetadata: boolean;
run: () => Promise<void>;
}
for (const tipo of ['emitidos', 'recibidos'] as const) {
try {
const result = await processDateRange(ctx, jobId, hace7Dias, ahora, tipo);
const makeStage = (
id: string,
label: string,
isMetadata: boolean,
runImpl: (stageId: string) => Promise<void>,
): DailyStage => ({ id, label, isMetadata, run: () => runImpl(id) });
const stages: DailyStage[] = [
makeStage('xml-emitidos-7d', 'XMLs emitidos últimos 7 días', false, async (id) => {
const result = await processDateRange(ctx, jobId, hace7Dias, ahora, 'emitidos', true, false, true);
totalFound += result.found;
totalDownloaded += result.downloaded;
totalInserted += result.inserted;
totalUpdated += result.updated;
} catch (error: any) {
console.error(`[SAT] Error XMLs ${tipo} (7 días):`, error.message);
if (result.errors.length > 0) {
for (const e of result.errors) nonFatalErrors.push({ stage: id, message: e.message });
}
}),
makeStage('xml-recibidos-7d', 'XMLs recibidos últimos 7 días', false, async (id) => {
const result = await processDateRange(ctx, jobId, hace7Dias, ahora, 'recibidos', true, false, true);
totalFound += result.found;
totalDownloaded += result.downloaded;
totalInserted += result.inserted;
totalUpdated += result.updated;
if (result.errors.length > 0) {
for (const e of result.errors) nonFatalErrors.push({ stage: id, message: e.message });
}
}),
];
// La metadata histórica (desde inicio de año) consume muchas solicitudes al SAT
// y no cambia día a día. Solo la ejecutamos los domingos para reducir carga.
if (ejecutarMetadataHistorica) {
const metaChunks = generateChunks(inicioAño, ahora, 3);
// Recorrer de más nuevo a más viejo: si un bloque antiguo da timeout, los
// recientes (más relevantes) ya quedaron procesados y no bloquean el avance.
for (let i = metaChunks.length - 1; i >= 0; i--) {
const { start, end } = metaChunks[i];
const chunkLabel = `${start.toISOString().slice(0, 10)}_${end.toISOString().slice(0, 10)}`;
stages.push(makeStage(
`metadata-emitidos-${chunkLabel}`,
`Metadata emitidos ${start.toISOString().slice(0, 10)} → ${end.toISOString().slice(0, 10)}`,
true,
async (id) => {
const { inserted, updated, errors } = await processMetadataRange(ctx, jobId, start, end, 'emitidos', false, true);
totalInserted += inserted;
totalUpdated += updated;
if (errors.length > 0) {
for (const e of errors) nonFatalErrors.push({ stage: id, message: e.message });
}
},
));
stages.push(makeStage(
`metadata-recibidos-${chunkLabel}`,
`Metadata recibidos ${start.toISOString().slice(0, 10)} → ${end.toISOString().slice(0, 10)}`,
true,
async (id) => {
const { inserted, updated, errors } = await processMetadataRange(ctx, jobId, start, end, 'recibidos', false, true);
totalInserted += inserted;
totalUpdated += updated;
if (errors.length > 0) {
for (const e of errors) nonFatalErrors.push({ stage: id, message: e.message });
}
},
));
}
}
// Paso 2: Metadata del ciclo fiscal actual (enero → hoy)
// Captura cancelaciones y cambios de status del año completo
console.log(`[SAT] Daily: Metadata desde ${inicioAño.toISOString().slice(0, 10)} → ${ahora.toISOString().slice(0, 10)}`);
const totalStages = stages.length;
let activeStageIndex = 0;
if (resumeFromStage) {
const idx = stages.findIndex(s => s.id === resumeFromStage);
if (idx >= 0) {
activeStageIndex = idx;
console.log(`[SAT Daily] Retomando desde etapa ${resumeFromStage} (${idx + 1}/${totalStages})`);
} else {
console.log(`[SAT Daily] No se encontró etapa ${resumeFromStage}, iniciando desde el principio`);
activeStageIndex = 0;
}
}
for (const tipo of ['emitidos', 'recibidos'] as const) {
const pendingStages: string[] = [];
const nonFatalErrors: { stage: string; message: string }[] = [];
for (let i = activeStageIndex; i < totalStages; i++) {
const stage = stages[i];
console.log(`[SAT Daily] Etapa ${i + 1}/${totalStages}: ${stage.label}`);
try {
const { inserted, updated } = await processMetadataRange(ctx, jobId, inicioAño, ahora, tipo);
totalInserted += inserted;
totalUpdated += updated;
await stage.run();
const progressPercent = totalStages > 0 ? Math.round(((i + 1) / totalStages) * 100) : 0;
await updateJobProgress(jobId, {
cfdisFound: totalFound,
cfdisDownloaded: totalDownloaded,
cfdisInserted: totalInserted,
cfdisUpdated: totalUpdated,
progressPercent,
});
} catch (error: any) {
console.error(`[SAT] Error metadata ${tipo} (ciclo fiscal):`, error.message);
const retryableErr = error instanceof SatSyncTimeoutError || error instanceof SatTransientError
? error
: (error.message?.includes('Timeout') ? new SatSyncTimeoutError(stage.id, error.message) : null);
// Errores transitorios (timeout, metadata aún no lista) se estacionan para
// reintento; no se pierden requestIds ya creados.
if (retryableErr && stage.isMetadata) {
console.warn(`[SAT Daily] Etapa ${stage.id} sin paquetes listos; se estaciona y se continúa con las demás.`);
pendingStages.push(stage.id);
continue;
}
if (retryableErr) {
throw retryableErr;
}
// Errores no transitorios (404 Error no controlado, etc.) no abortan el
// daily. Se registran para diagnóstico y se continúa con las demás etapas.
console.error(`[SAT Daily] Etapa ${stage.id} falló (no transitorio): ${error.message}`);
nonFatalErrors.push({ stage: stage.id, message: error.message });
continue;
}
}
const errorMessage = nonFatalErrors.length > 0
? JSON.stringify({ completedWithWarnings: true, errors: nonFatalErrors })
: undefined;
await updateJobProgress(jobId, {
cfdisFound: totalFound,
cfdisDownloaded: totalDownloaded,
cfdisInserted: totalInserted,
cfdisUpdated: totalUpdated,
progressPercent: 100,
errorMessage,
});
if (pendingStages.length > 0) {
throw new SatMetadataPendingError(
pendingStages[0],
pendingStages,
`Metadata no lista en ${pendingStages.length} chunk(s): ${pendingStages.join(', ')}`,
);
}
}
/**
* Verifica y descarga un requestId existente, procesando los paquetes resultantes.
* Similar a processMetadataRange pero recibiendo el requestId explícito.
*/
async function requestAndDownloadWithId(
ctx: SyncContext,
jobId: string,
requestId: string,
tipoCfdi: 'emitidos' | 'recibidos',
requestType: 'metadata',
fechaInicio: Date,
fechaFin: Date,
isDaily = false,
): Promise<{ inserted: number; updated: number }> {
let totalInserted = 0;
let totalUpdated = 0;
const label = `${tipoCfdi}/${requestType} ${fechaInicio.toISOString().slice(0, 10)} → ${fechaFin.toISOString().slice(0, 10)}`;
let verifyResult: Awaited<ReturnType<typeof verifySatRequest>> | undefined;
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));
attempts++;
verifyResult = await verifySatRequest(ctx.service, requestId);
console.log(`[SAT] Estado ${label}: ${verifyResult.status} (intento ${attempts}/${maxAttempts})`);
if (verifyResult.status === 'ready') break;
if (verifyResult.status === 'failed' || verifyResult.status === 'rejected') {
if (isAgotadas(verifyResult.message)) {
console.log(`[SAT] Solicitudes agotadas de por vida (${label}); se cancela y se omite.`);
return { inserted: 0, updated: 0 };
}
throw new Error(`Solicitud fallida (${label}): ${verifyResult.message}`);
}
}
if (!verifyResult || verifyResult.status !== 'ready') {
throw new SatSyncTimeoutError(stageIdForTimeout(label), `Timeout esperando respuesta del SAT (${label})`);
}
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);
if (!downloadResult.success) {
console.error(`[SAT] Error descargando paquete ${packageId}: ${downloadResult.message}`);
continue;
}
const items = processMetadataPackage(downloadResult.packageContent, tipoCfdi);
console.log(`[SAT] Procesando ${items.length} registros de metadata ${tipoCfdi}`);
const { inserted, updated } = await saveMetadata(await ctx.getPool(), items, jobId, ctx.contribuyenteId);
totalInserted += inserted;
totalUpdated += updated;
}
return { inserted: totalInserted, updated: totalUpdated };
}
/**
@@ -1294,29 +1600,54 @@ export async function startSync(
} catch (error: any) {
console.error(`[SAT] Error en sincronización ${job.id}:`, error);
const isTimeout = error.message?.includes('Timeout');
const isMetadataPending = error instanceof SatMetadataPendingError;
const isSatTimeout = error instanceof SatSyncTimeoutError;
const isTransient = error instanceof SatTransientError;
const isTimeout = isSatTimeout || isMetadataPending || isTransient || error.message?.includes('Timeout');
const currentRetries = job.retryCount || 0;
const policy = getRetryPolicy(job);
const nextRetryNumber = currentRetries + 1;
const nextRetry = isTimeout && nextRetryNumber <= policy.maxRetries
? computeNextRetryAt(job.startedAt!, nextRetryNumber, policy)
// Para daily permitimos hasta MAX_DAILY_RETRY_ATTEMPTS intentos totales.
// Los primeros usan la política automática (6h/12h); los restantes los
// recogen los crons fijos de 9 AM / 4 PM CDMX.
const maxAttempts = job.type === 'daily' ? MAX_DAILY_RETRY_ATTEMPTS : policy.maxRetries;
const hasAttemptsLeft = isTimeout && nextRetryNumber <= maxAttempts;
// B: anclar la política a createdAt (inmutable) en vez de startedAt,
// porque startedAt se resetea en cada retry para que el watchdog mida
// el intento actual y no mate retries legítimos en vuelo.
const nextRetry = hasAttemptsLeft
? (computeNextRetryAt(job.createdAt, nextRetryNumber, policy) ?? null)
: null;
if (nextRetry) {
// Para timeouts/rechazos transitorios del daily, persistimos la etapa para retomar en retries programados.
const progressErrorMessage = isMetadataPending
? JSON.stringify({ stage: error.stageId, pendingStages: error.pendingStages, message: error.message })
: isSatTimeout || isTransient
? JSON.stringify({ stage: error.stageId, message: error.message })
: undefined;
if (hasAttemptsLeft) {
const retryLabel = nextRetry
? nextRetry.toLocaleString('es-MX')
: 'próxima ventana 9 AM / 4 PM CDMX';
await updateJobProgress(job.id, {
status: 'pending',
errorMessage: `Timeout (intento ${nextRetryNumber}/${policy.maxRetries}). Reintento programado para ${nextRetry.toLocaleString('es-MX')}.`,
errorMessage: progressErrorMessage ?? `Timeout (intento ${nextRetryNumber}/${maxAttempts}). Reintento programado para ${retryLabel}.`,
retryCount: nextRetryNumber,
nextRetryAt: nextRetry,
nextRetryAt: nextRetry ?? null as any,
});
console.log(`[SAT] Job ${job.id} programado para reintento ${nextRetryNumber}/${policy.maxRetries} a las ${nextRetry.toLocaleString('es-MX')}`);
console.log(`[SAT] Job ${job.id} programado para reintento ${nextRetryNumber}/${maxAttempts} (${retryLabel})`);
} else {
// Sin reintentos restantes, error no-timeout, o policy con maxRetries=0 (incremental)
const finalMsg = isTimeout
? policy.maxRetries === 0
? 'Timeout en sync incremental — sin reintentos por política. Próximo cron incremental cubrirá el gap.'
: 'Fallo conexión SAT, vuelve a intentar con un rango de fechas menor.'
: error.message;
const finalMsg = isMetadataPending
? progressErrorMessage
: isSatTimeout
? progressErrorMessage
: isTimeout
? policy.maxRetries === 0
? 'Timeout en sync incremental — sin reintentos por política. Próximo cron incremental cubrirá el gap.'
: 'Fallo conexión SAT, vuelve a intentar con un rango de fechas menor.'
: error.message;
await updateJobProgress(job.id, {
status: 'failed',
errorMessage: finalMsg,
@@ -1410,7 +1741,23 @@ export async function retryTimedOutJobs(): Promise<void> {
getPool: () => tenantDb.getPool(job.tenantId, job.tenant.databaseName),
};
await updateJobProgress(job.id, { status: 'running', errorMessage: null as any });
// B: resetear startedAt al inicio de ESTE intento para que el watchdog
// mida el intento actual (y no mate un retry legítimo por el startedAt
// original del job). La política de retries se ancla a createdAt.
await updateJobProgress(job.id, { status: 'running', errorMessage: null as any, startedAt: new Date() });
// Para jobs daily, intentamos retomar desde la última etapa completada.
let resumeFromStage: string | undefined;
if (job.type === 'daily' && job.errorMessage) {
try {
const parsed = JSON.parse(job.errorMessage);
if (typeof parsed.stage === 'string') {
resumeFromStage = parsed.stage;
}
} catch {
// errorMessage no es JSON, ignorar
}
}
// Re-ejecutar según tipo original
try {
@@ -1419,7 +1766,7 @@ export async function retryTimedOutJobs(): Promise<void> {
} else if (job.type === 'incremental') {
await processIncrementalSync(ctx, job.id);
} else {
await processDailySync(ctx, job.id);
await processDailySync(ctx, job.id, resumeFromStage);
}
await updateJobProgress(job.id, {
@@ -1432,26 +1779,46 @@ export async function retryTimedOutJobs(): Promise<void> {
} catch (retryError: any) {
console.error(`[SAT Retry] Job ${job.id} falló de nuevo:`, retryError.message);
const isTimeout = retryError.message?.includes('Timeout');
const isMetadataPending = retryError instanceof SatMetadataPendingError;
const isSatTimeout = retryError instanceof SatSyncTimeoutError;
const isTransient = retryError instanceof SatTransientError;
const isTimeout = isSatTimeout || isMetadataPending || isTransient || retryError.message?.includes('Timeout');
const policy = getRetryPolicy(job);
const nextRetryNumber = job.retryCount + 1;
const nextRetry = isTimeout && nextRetryNumber <= policy.maxRetries
? computeNextRetryAt(job.startedAt!, nextRetryNumber, policy)
// Para daily permitimos hasta MAX_DAILY_RETRY_ATTEMPTS intentos totales.
const maxAttempts = job.type === 'daily' ? MAX_DAILY_RETRY_ATTEMPTS : policy.maxRetries;
const hasAttemptsLeft = isTimeout && nextRetryNumber <= maxAttempts;
// B: política anclada a createdAt (startedAt se resetea por intento).
const nextRetry = hasAttemptsLeft
? (computeNextRetryAt(job.createdAt, nextRetryNumber, policy) ?? null)
: null;
if (nextRetry) {
const progressErrorMessage = isMetadataPending
? JSON.stringify({ stage: retryError.stageId, pendingStages: retryError.pendingStages, message: retryError.message })
: isSatTimeout || isTransient
? JSON.stringify({ stage: retryError.stageId, message: retryError.message })
: undefined;
if (hasAttemptsLeft) {
const retryLabel = nextRetry
? nextRetry.toLocaleString('es-MX')
: 'próxima ventana 9 AM / 4 PM CDMX';
await updateJobProgress(job.id, {
status: 'pending',
errorMessage: `Timeout (intento ${nextRetryNumber}/${policy.maxRetries}). Reintento programado para ${nextRetry.toLocaleString('es-MX')}.`,
errorMessage: progressErrorMessage ?? `Timeout (intento ${nextRetryNumber}/${maxAttempts}). Reintento programado para ${retryLabel}.`,
retryCount: nextRetryNumber,
nextRetryAt: nextRetry,
nextRetryAt: nextRetry ?? null as any,
});
} else {
await updateJobProgress(job.id, {
status: 'failed',
errorMessage: isTimeout
? 'Fallo conexión SAT, vuelve a intentar con un rango de fechas menor.'
: retryError.message,
errorMessage: isMetadataPending
? progressErrorMessage
: isSatTimeout
? progressErrorMessage
: isTimeout
? 'Fallo conexión SAT, vuelve a intentar con un rango de fechas menor.'
: retryError.message,
completedAt: new Date(),
});
}
@@ -1467,6 +1834,153 @@ export async function retryTimedOutJobs(): Promise<void> {
}
}
/**
* Retoma jobs diarios que quedaron pending por timeout de polling, sin depender
* de nextRetryAt. Usado por los cron fijos de 9:00 AM y 4:00 PM CDMX.
* Hasta MAX_DAILY_RETRY_ATTEMPTS intentos en total.
*/
export async function continuePendingDailyRequests(): Promise<void> {
const pendingJobs = await prisma.satSyncJob.findMany({
where: {
status: 'pending',
type: 'daily',
retryCount: { lt: MAX_DAILY_RETRY_ATTEMPTS },
},
include: { tenant: { select: { id: true, databaseName: true, rfc: true } } },
});
if (pendingJobs.length === 0) {
console.log('[SAT Daily Retry] No hay jobs diarios pendientes');
return;
}
console.log(`[SAT Daily Retry] ${pendingJobs.length} job(s) diarios pendientes`);
for (const job of pendingJobs) {
try {
const activeSync = await prisma.satSyncJob.findFirst({
where: {
tenantId: job.tenantId,
contribuyenteId: job.contribuyenteId ?? null,
status: 'running',
},
});
if (activeSync) {
console.log(`[SAT Daily Retry] (${job.tenant.rfc}, contrib=${job.contribuyenteId || 'tenant-wide'}) tiene sync activo, posponiendo`);
continue;
}
console.log(`[SAT Daily Retry] Reintentando job ${job.id} (${job.tenant.rfc}), intento ${(job.retryCount || 0) + 1}/${MAX_DAILY_RETRY_ATTEMPTS}`);
let decryptedFiel = null;
if (job.contribuyenteId) {
const pool = await tenantDb.getPool(job.tenantId, job.tenant.databaseName);
decryptedFiel = await getDecryptedFielContribuyente(pool, job.contribuyenteId);
}
if (!decryptedFiel) {
decryptedFiel = await getDecryptedFiel(job.tenantId);
}
if (!decryptedFiel) {
await updateJobProgress(job.id, {
status: 'failed',
errorMessage: 'FIEL no disponible para reintento',
completedAt: new Date(),
});
continue;
}
const service = createSatService({
cerContent: decryptedFiel.cerContent,
keyContent: decryptedFiel.keyContent,
password: decryptedFiel.password,
});
const ctx: SyncContext = {
fielData: {
cerContent: decryptedFiel.cerContent,
keyContent: decryptedFiel.keyContent,
password: decryptedFiel.password,
},
service,
rfc: decryptedFiel.rfc,
tenantId: job.tenantId,
databaseName: job.tenant.databaseName,
contribuyenteId: job.contribuyenteId ?? null,
getPool: () => tenantDb.getPool(job.tenantId, job.tenant.databaseName),
};
let resumeFromStage: string | undefined;
if (job.errorMessage) {
try {
const parsed = JSON.parse(job.errorMessage);
if (typeof parsed.stage === 'string') {
resumeFromStage = parsed.stage;
}
} catch {
// no es JSON, ignorar
}
}
// B: resetear startedAt al inicio de este intento (ver retryTimedOutJobs).
await updateJobProgress(job.id, { status: 'running', errorMessage: null as any, startedAt: new Date() });
try {
await processDailySync(ctx, job.id, resumeFromStage);
await updateJobProgress(job.id, {
status: 'completed',
completedAt: new Date(),
progressPercent: 100,
errorMessage: null as any,
});
console.log(`[SAT Daily Retry] Job ${job.id} completado`);
} catch (retryError: any) {
console.error(`[SAT Daily Retry] Job ${job.id} falló:`, retryError.message);
const isMetadataPending = retryError instanceof SatMetadataPendingError;
const isSatTimeout = retryError instanceof SatSyncTimeoutError;
const isTransient = retryError instanceof SatTransientError;
const isTimeout = isSatTimeout || isMetadataPending || isTransient || retryError.message?.includes('Timeout');
const nextRetryCount = (job.retryCount || 0) + 1;
const progressErrorMessage = isMetadataPending
? JSON.stringify({ stage: retryError.stageId, pendingStages: retryError.pendingStages, message: retryError.message })
: isSatTimeout || isTransient
? JSON.stringify({ stage: retryError.stageId, message: retryError.message })
: undefined;
if (isTimeout && nextRetryCount < MAX_DAILY_RETRY_ATTEMPTS) {
await updateJobProgress(job.id, {
status: 'pending',
errorMessage: progressErrorMessage,
retryCount: nextRetryCount,
nextRetryAt: null as any,
});
console.log(`[SAT Daily Retry] Job ${job.id} quedó pending para siguiente ventana (intento ${nextRetryCount}/${MAX_DAILY_RETRY_ATTEMPTS})`);
} else {
await updateJobProgress(job.id, {
status: 'failed',
errorMessage: isMetadataPending
? progressErrorMessage
: isSatTimeout
? progressErrorMessage
: isTimeout
? 'Fallo conexión SAT, vuelve a intentar con un rango de fechas menor.'
: retryError.message,
completedAt: new Date(),
});
}
}
} catch (error: any) {
console.error(`[SAT Daily Retry] Error procesando job ${job.id}:`, error.message);
await updateJobProgress(job.id, {
status: 'failed',
errorMessage: error.message,
completedAt: new Date(),
});
}
}
}
/**
* Obtiene el estado actual de sincronización de un tenant
*/