rfidnatura/worker/worker.js

const { parentPort } = require('worker_threads');
const fs = require('fs').promises;
const path = require('path');
const pool = require('../db/pool');
const log4js = require('log4js');
const dotenv = require('dotenv');
const net = require('net');
const generateZPL = require('../resources/zplTemplate');
const label = require('../resources/label.json');

dotenv.config();
// Configuración de log4js
log4js.configure({
    appenders: {
        file: { type: 'file', filename: 'logs/app.log', maxLogSize: 10485760, backups: 3, compress: true },
        console: { type: 'console' }
    },
    categories: {
        default: { appenders: ['file', 'console'], level: 'info' }
    }
});

const logger = log4js.getLogger('default');

// Función para obtener una conexión con reintentos
async function getConnectionWithRetries(retries = 3, delay = 5000) {
    let attempt = 0;
    while (attempt < retries) {
        try {
            return await pool.getConnection(); 
        } catch (error) {
            attempt++;
            if (attempt >= retries)
                throw new Error('No se pudo obtener conexión después de múltiples intentos.');

            await new Promise(resolve => setTimeout(resolve, delay));
        }
    }
}

//Función para enviar ZPL a la impresora con TCP socket y reintentos
async function sendZPL(zplContent, ipAddress, retries = 2) {
    const port = 9100;
    try {
        const result = await new Promise((resolve, reject) => {
            const client = new net.Socket();

            client.setTimeout(1000);

            client.connect(port, ipAddress, () => {
                client.write(zplContent, () => {
                    client.destroy();
                    resolve('success');
                });
            });

            client.on('timeout', () => {
                logger.error('Tiempo de espera agotado al conectar con la impresora');
                client.destroy();
                reject('error');
            });

            client.on('error', error => {
                logger.error(`Error al enviar ZPL a la impresora: ${error.message || error}`);

                client.destroy();
                reject('error');
            });
        });

        if (result === 'success') return result;
    } catch (error) {
        logger.error(`Error al enviar ZPL a la impresora: ${error.message || error}`);
        if (retries > 0) {
            logger.info(`Reintentando... Quedan ${retries} intentos.`);
            return sendZPL(zplContent, ipAddress, retries - 1);
        } else {
            logger.error('Todos los intentos fallaron. No se pudo enviar el ZPL.');
            return 'error';
        }
    }
}

// Función para obtener la dirección IP de la impresora, realizando una llamada a la base de datos a la tabla de printer
async function getPrinterIpAddress(printerFk, connection) {
    try {
        const [rows] = await connection.query(`
            SELECT ipAddress
                FROM ${process.env.DB_PRINTER_SCHEMA}.printer
                WHERE id = ?
        `, [printerFk]);
        if (!rows.length)
            throw new Error(`No se encontró la impresora con id = ${printerFk}`);
        return rows[0].ipAddress;
    } catch (error) {
        logger.error('Error al obtener la dirección IP de la impresora:', error);
        throw error;
    }
}

// Función para obtener un único registro para procesar
async function getRecordForProcessing(retries = 5, delay = 4000) {
    for (let attempt = 0; attempt < retries; attempt++) {
        const conn = await getConnectionWithRetries();
        try {
            await conn.beginTransaction();
            const [rows] = await conn.query(`
                SELECT * FROM expedition_PrintOut
                    WHERE isPrinted = 10
                    LIMIT 1 FOR UPDATE
            `);
            if (!rows.length) {
                await conn.commit();
                return;
            }

            const record = rows[0];
            await conn.query(`
                UPDATE expedition_PrintOut
                    SET isPrinted = 12
                    WHERE expeditionFk = ?
             `, [record.expeditionFk]);
            await conn.commit();
            return record;
        } catch (error) {
            await conn.rollback();
            if (error.code === 'ER_LOCK_WAIT_TIMEOUT') {
                logger.error('Lock wait timeout exceeded, retrying...');
                if (attempt >= retries - 1)
                    throw new Error('No se pudo obtener el registro después de múltiples intentos.');
                await new Promise(resolve => setTimeout(resolve, delay));
            } else {
                logger.error('Error al obtener y marcar el registro para procesamiento:', error);
                return;
            }
        } finally {
            conn.release();
        }
    }
}

// Función para procesar un único registro
async function processRecord(record) {
    const connection = await getConnectionWithRetries();

    try {
        await connection.beginTransaction();

        logger.info(`(${record.expeditionFk}) Procesando...`);
        const zplData = generateZPL(record, label);

        const filePath = path.join(__dirname, `zplData_${record.expeditionFk}.txt`);

        await fs.writeFile(filePath, zplData["VerdNatura Label RFID"].zpl, 'utf8');

        const zplContent = await fs.readFile(filePath, 'utf8');
        const ipAddress = await getPrinterIpAddress(record.printerFk, connection);

        const sendResult = await sendZPL(zplContent, ipAddress);

        fs.unlink(filePath);

        if (sendResult === 'success') {
            await connection.query(`
                UPDATE expedition_PrintOut
                    SET isPrinted = 11
                    WHERE expeditionFk = ?
            `, [record.expeditionFk]);
            logger.info(`(${record.expeditionFk}) Base de datos actualizada`);
        } else {
            await connection.query(`
                UPDATE expedition_PrintOut
                    SET isPrinted = 13
                    WHERE expeditionFk = ?
            `, [record.expeditionFk]);
            logger.error(`(${record.expeditionFk}) Error al enviar ZPL a la impresora`);
        }
        parentPort.postMessage('done');
        await connection.commit();
    } catch (error) {
        logger.error('Error al procesar el registro:', error);
        parentPort.postMessage('error');
        // if (!error.message === `Can't add new command when connection is in closed state`)
        await connection.rollback();
    } finally {
        connection.release();
    }
}

// Escuchar mensajes del hilo principal
parentPort.on('message', async msg => {
    if (msg === 'check') {
        const record = await getRecordForProcessing();
        if (record) 
            await processRecord(record);
        else {
            // Si no hay registros, espera y vuelve a verificar
            setTimeout(async() => {
                parentPort.postMessage('done');
            }, 5000);
        }
    } else
        processRecord(msg).catch(err => logger.error('Error en el worker:', err));
});