HoruxDespachosNuevo/apps/api/scripts/sync-daily-cyl.ts

import { startSync } from '../src/services/sat/sat.service.js';
import { prisma, tenantDb } from '../src/config/database.js';

const TENANT_ID = '49b60455-c501-4ca2-b4bc-36ea7f2951a2';
const CONCURRENCY = 3;
const POLL_INTERVAL_MS = 60_000;

function sleep(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

async function waitForActiveSync(tenantId: string, contribuyenteId: string): Promise<'clean' | 'pending-retry'> {
  const active = await prisma.satSyncJob.findFirst({
    where: {
      tenantId,
      contribuyenteId,
      status: { in: ['pending', 'running'] },
    },
    orderBy: { createdAt: 'desc' },
  });
  if (!active) return 'clean';

  if (active.status === 'pending') {
    console.log(`[SKIP] Contribuyente ${contribuyenteId} tiene reintento programado (${active.id}); se omite en este ciclo.`);
    return 'pending-retry';
  }

  while (true) {
    const running = await prisma.satSyncJob.findFirst({
      where: {
        tenantId,
        contribuyenteId,
        status: 'running',
      },
    });
    if (!running) return 'clean';
    console.log(`[WAIT] Sync activo para ${contribuyenteId}. Esperando 60s...`);
    await sleep(POLL_INTERVAL_MS);
  }
}

async function waitForJob(jobId: string): Promise<{ status: string; error: string | null; found: number; inserted: number; updated: number; progress: number }> {
  while (true) {
    const job = await prisma.satSyncJob.findUnique({ where: { id: jobId } });
    if (!job) throw new Error(`Job ${jobId} no encontrado`);
    // Un job que queda `pending` con nextRetryAt fue un fallo transitorio; el
    // retry automático lo retomará. No bloquear el batch manual esperando horas.
    if (job.status === 'completed' || job.status === 'failed' || job.status === 'pending') {
      return {
        status: job.status,
        error: job.errorMessage,
        found: job.cfdisFound || 0,
        inserted: job.cfdisInserted || 0,
        updated: job.cfdisUpdated || 0,
        progress: job.progressPercent || 0,
      };
    }
    console.log(`[WAIT] Job ${jobId} status=${job.status}, progress=${job.progressPercent}%. Esperando 60s...`);
    await sleep(POLL_INTERVAL_MS);
  }
}

async function runWithConcurrency<T>(items: T[], concurrency: number, fn: (item: T) => Promise<void>): Promise<void> {
  let index = 0;

  async function worker() {
    while (index < items.length) {
      const item = items[index++];
      await fn(item);
    }
  }

  const workers = Array.from({ length: concurrency }, () => worker());
  await Promise.all(workers);
}

async function main() {
  const tenant = await prisma.tenant.findUnique({
    where: { id: TENANT_ID },
    select: { id: true, rfc: true, nombre: true, databaseName: true },
  });

  if (!tenant) {
    console.error(`[C&L Daily] Tenant ${TENANT_ID} no encontrado`);
    process.exit(1);
  }

  console.log(`[C&L Daily] Tenant: ${tenant.nombre} (${tenant.rfc}) | DB: ${tenant.databaseName}`);

  const pool = await tenantDb.getPool(tenant.id, tenant.databaseName);
  const { rows: contribuyentes } = await pool.query(`
    SELECT c.entidad_id AS id, c.rfc, eg.nombre
    FROM contribuyentes c
    JOIN entidades_gestionadas eg ON eg.id = c.entidad_id
    JOIN fiel_contribuyente f ON f.contribuyente_id = c.entidad_id
    WHERE f.is_active = true
      AND f.valid_until >= NOW()
    ORDER BY eg.nombre
  `);

  console.log(`[C&L Daily] Contribuyentes con FIEL activa: ${contribuyentes.length}`);

  const errors: string[] = [];

  await runWithConcurrency(contribuyentes, CONCURRENCY, async (c: any) => {
    console.log(`\n[SYNC] === ${c.rfc} | ${c.nombre} ===`);
    try {
      const activeState = await waitForActiveSync(tenant.id, c.id);
      if (activeState === 'pending-retry') {
        return;
      }

      const jobId = await startSync(tenant.id, 'daily', undefined, undefined, c.id);
      console.log(`[SYNC] Job iniciado: ${jobId} para ${c.rfc}`);

      const result = await waitForJob(jobId);
      console.log(`[SYNC] ${c.rfc} finalizado: status=${result.status}, found=${result.found}, inserted=${result.inserted}, updated=${result.updated}, progress=${result.progress}%`);
      if (result.error) {
        console.error(`[SYNC] Error en ${c.rfc}: ${result.error}`);
        errors.push(`${c.rfc}: ${result.error}`);
      }
    } catch (error: any) {
      console.error(`[SYNC] Error lanzando sync para ${c.rfc}:`, error.message || error);
      errors.push(`${c.rfc}: ${error.message || error}`);
    }
  });

  console.log('\n[C&L Daily] Proceso finalizado');
  if (errors.length > 0) {
    console.error(`[C&L Daily] Errores (${errors.length}):`);
    for (const e of errors) console.error(`  - ${e}`);
    process.exit(1);
  }
}

main().catch(async (err) => {
  console.error('[C&L Daily] Error fatal:', err);
  await prisma.$disconnect().catch(() => {});
  process.exit(1);
});