import { AppError } from '../utils/errors.js' import { prismaStateStore, type StateStore } from './automatizaciones.service.js' import { mondayService, type MondayService } from './monday.service.js' import { prisma } from './prisma.js' import { saludMonday, type SaludMondayService } from './salud-monday.service.js' /** * Sync worker del modo local: drena la cola SyncPendiente hacia Monday. * * - Solo corre cuando Monday está disponible (saludMonday); si está caído es no-op. * - FIFO por createdAt (la captura encoló en orden: lead → conversión → * afluencia → apartado → contacto) y SECUENCIAL: SQLite serializa las * escrituras (P1008) si se procesa en paralelo. * - Idempotencia con AutomatizacionState, clave `sync:` (mismo patrón que * `apartado-creado:` de automatizaciones): si el proceso muere a media * pasada, al reintentar no duplica el item en Monday. * - Fallo Monday (AppError 502): vuelve a PENDIENTE con intentos+1; al superar * MAX_INTENTOS queda ERROR (visible en /api/sistema/estado). Cualquier otro * error (tipo sin handler, payload inválido) marca ERROR de inmediato. * * Los handlers por tipo los registra la fase B (con los payloads reales de * cada flujo) vía `registrarHandler`; el mapa queda extensible y defensivo. */ export const TIPOS_SYNC = [ 'CREAR_LEAD', 'CREAR_AFLUENCIA', 'ACTUALIZAR_AFLUENCIA', 'CAMBIO_ETAPA', 'DESCARTE', 'CONVERSION', 'CREAR_CONTACTO', 'ACTUALIZAR_CONTACTO', 'CREAR_APARTADO', 'COTIZACION_OPORTUNIDAD', 'CHECKBOX_DOC', 'APARTO_OPORTUNIDAD', 'LIQUIDADO_ESCRITURAR', 'ASIGNAR_PERSONA', 'DISPONIBILIDAD_UNIDAD', ] as const export type TipoSync = (typeof TIPOS_SYNC)[number] /** Intentos máximos antes de declarar ERROR definitivo */ export const MAX_INTENTOS_SYNC = 10 /** Registro de la cola tal como lo consume el worker */ export interface SyncRegistro { id: string tipo: string payloadJson: string referenciaLocal: string | null estado: string intentos: number createdAt: Date } /** Acceso mínimo a la tabla SyncPendiente (Prisma en prod, memoria en tests) */ export interface SyncStore { /** Hasta `limite` registros PENDIENTE en orden FIFO (createdAt asc) */ pendientes(limite: number): Promise marcar( id: string, datos: { estado: string; intentos?: number; ultimoError?: string | null }, ): Promise } export interface ContextoSync { monday: MondayService referenciaLocal: string | null } /** * Ejecuta la réplica en Monday. Devuelve el id creado (o cualquier cadena) para * guardarlo en el dedup `sync:`; void se persiste como 'ok'. */ export type HandlerSync = ( payload: Record, ctx: ContextoSync, ) => Promise export interface ResultadoSync { procesados: number fallos: number } /** true si el error es un fallo de Monday retryable (AppError 502 de graphql) */ export function esFalloMonday(err: unknown): boolean { return err instanceof AppError && err.status === 502 } /** * El handler no puede ejecutarse TODAVÍA porque una operación previa de la * cola aún no se sincroniza (p.ej. CAMBIO_ETAPA sobre un item cuyo CREATE no * ha corrido: no hay mondayItemId). El worker lo deja PENDIENTE SIN sumar * intento: se reintenta en la siguiente pasada, cuando su dependencia ya se * haya aplicado (la cola es FIFO). */ export class DependenciaPendienteError extends Error { constructor(message: string) { super(message) this.name = 'DependenciaPendienteError' } } /** Clave de dedup/idempotencia del registro en AutomatizacionState */ export function claveSync(id: string): string { return `sync:${id}` } export class SyncMondayService { private readonly handlers = new Map() /** Mutex en memoria: evita solapar el cron de 2 min con el disparo al recuperar salud */ private enEjecucion = false constructor( private readonly monday: MondayService, private readonly state: StateStore, private readonly db: SyncStore, private readonly salud: Pick = saludMonday, ) {} /** Registra el handler de un tipo de réplica (lo usa la fase B por flujo) */ registrarHandler(tipo: TipoSync, handler: HandlerSync): void { this.handlers.set(tipo, handler) } /** * Una pasada por la cola (máx `limite` registros). No-op si Monday está * caído / MODO_LOCAL_FORZADO, o si ya hay una pasada en curso. */ async procesarPendientes({ limite = 25 }: { limite?: number } = {}): Promise { if (!this.salud.estaDisponible()) return { procesados: 0, fallos: 0 } if (this.enEjecucion) return { procesados: 0, fallos: 0 } this.enEjecucion = true try { const pendientes = await this.db.pendientes(limite) let procesados = 0 let fallos = 0 for (const registro of pendientes) { await this.db.marcar(registro.id, { estado: 'PROCESANDO' }) try { await this.ejecutar(registro) await this.db.marcar(registro.id, { estado: 'HECHO', ultimoError: null }) procesados++ } catch (err) { const mensaje = err instanceof Error ? err.message : String(err) if (err instanceof DependenciaPendienteError) { // Falta una operación previa de la cola: queda PENDIENTE sin sumar // intento (no es un fallo; se reintenta en la próxima pasada) await this.db.marcar(registro.id, { estado: 'PENDIENTE', ultimoError: mensaje }) } else { fallos++ const intentos = registro.intentos + 1 if (esFalloMonday(err)) { // Monday volvió a caer a media pasada: reintenta en la siguiente await this.db.marcar(registro.id, { estado: intentos > MAX_INTENTOS_SYNC ? 'ERROR' : 'PENDIENTE', intentos, ultimoError: mensaje, }) } else { // Error de datos/código: reintentar no lo arregla await this.db.marcar(registro.id, { estado: 'ERROR', intentos, ultimoError: mensaje }) } } } } if (procesados + fallos > 0) { // eslint-disable-next-line no-console console.log(`[sync] procesados ${procesados}, fallos ${fallos}`) } return { procesados, fallos } } finally { this.enEjecucion = false } } /** Ejecuta un registro con dedup `sync:`; lanza para que el caller clasifique */ private async ejecutar(registro: SyncRegistro): Promise { const handler = this.handlers.get(registro.tipo) if (!handler) throw new Error(`Tipo de sync sin handler registrado: "${registro.tipo}"`) let payload: Record try { payload = JSON.parse(registro.payloadJson) as Record } catch { throw new Error(`payloadJson inválido (no es JSON): registro ${registro.id}`) } const clave = claveSync(registro.id) const previo = await this.state.get(clave) if (previo !== null) return // ya se aplicó en una pasada previa (dedup) const resultado = await handler(payload, { monday: this.monday, referenciaLocal: registro.referenciaLocal, }) await this.state.set(clave, typeof resultado === 'string' && resultado !== '' ? resultado : 'ok') } } /** Store respaldado por la tabla SyncPendiente de Prisma */ export const prismaSyncStore: SyncStore = { pendientes(limite) { return prisma.syncPendiente.findMany({ where: { estado: 'PENDIENTE' }, orderBy: { createdAt: 'asc' }, take: limite, }) }, async marcar(id, datos) { await prisma.syncPendiente.update({ where: { id }, data: datos }) }, } /** Singleton usado por los crons de index.ts (los tests instancian el suyo) */ export const syncMondayService = new SyncMondayService(mondayService, prismaStateStore, prismaSyncStore)