BullMQ Queue System

const redisUrl = process.env.REDIS_URL;
const redisOpts = { maxRetriesPerRequest: null };

if (!redisUrl) {
    redisOpts.host = process.env.REDIS_HOST || 'localhost';
    redisOpts.port = process.env.REDIS_PORT || 6379;
    redisOpts.password = process.env.REDIS_PASSWORD;
}

const queueConnection = redisUrl ? new IORedis(redisUrl, redisOpts) : new IORedis(redisOpts);
const workerConnection = redisUrl ? new IORedis(redisUrl, redisOpts) : new IORedis(redisOpts);

export const whatsappQueue = new Queue('whatsapp-ai', { connection: queueConnection });

await whatsappQueue.add('process-message', {
    wamid,
    from: message.from,
    text: message.text.body,
    timestamp: message.timestamp
}, {
    jobId: wamid, // Deduplication via BullMQ
    removeOnComplete: true
});

export const worker = new Worker('whatsapp-ai', async job => {
    const { wamid, from, text } = job.data;
    console.log(`⚙️ Processing Job: ${job.id} - ${text}`);

    try {
        // Processing logic...
    } catch (error) {
        console.error(`❌ Job Failed: ${error.message}`);
        throw error;
    }
}, { 
    connection: workerConnection,
    limiter: {
        max: 10, // Max 10 processing jobs at a time
        duration: 1000
    },
    settings: {
        backoffStrategy: 'exponential'
    }
});

settings: {
    backoffStrategy: 'exponential'
}

let session = await prisma.whatsAppSession.findUnique({ where: { phoneNumber: from } });

if (!session) {
    session = await prisma.whatsAppSession.create({
        data: { 
            phoneNumber: from, 
            isBotActive: true, 
            expiresAt: new Date(Date.now() + 24 * 60 * 60 * 1000),
            sessionContext: { history: [] } 
        }
    });
    // Emit New Session
    if (io) io.emit('session_new', { id: session.id, phone: from, time: session.updatedAt });
}

if (!session.isBotActive) {
    console.log(`⏸️ Bot inactive for ${from}. Skipping.`);
    return;
}

const cleanText = redactPII(text);

const userMsg = { role: 'user', content: cleanText };
history.push(userMsg);

if (io) {
    io.to(`session_${session.id}`).emit('new_message', { 
        sessionId: session.id, 
        message: { role: 'user', content: text, time: "Just now" } 
    });
    io.emit('chat_list_update', { sessionId: session.id });
}

const HANDOVER_KEYWORDS = /\b(humano|agente|asesor|persona|queja|reclamo|ayuda|contactar|hablar con alguien)\b/i;

if (HANDOVER_KEYWORDS.test(text)) {
    console.log(`🚨 Handover triggered for ${from} by text: "${text}"`);
    
    await prisma.whatsAppSession.update({
        where: { id: session.id },
        data: { 
            isBotActive: false,
            handoverTrigger: "KEYWORD_DETECTED",
            sessionContext: { ...session.sessionContext, history }
        }
    });

    if (io) io.emit('session_update', { id: session.id, status: 'handover' });

    await axios.post(
        `https://graph.facebook.com/v21.0/${process.env.WHATSAPP_PHONE_ID}/messages`,
        {
            messaging_product: "whatsapp",
            to: from,
            text: { body: "Te estoy transfiriendo con un asesor humano. Un momento por favor." }
        },
        { headers: { 'Authorization': `Bearer ${process.env.WHATSAPP_ACCESS_TOKEN}`, 'Content-Type': 'application/json' } }
    );
    
    return; // STOP AI PROCESSING
}

const aiResponse = await generateSupportResponse(text, history);

let finalText = aiResponse.text;
const imageRegex = /\[SEND_IMAGE:\s*([^\]]+)\]/g;
let match;
const imageIds = [];

while ((match = imageRegex.exec(finalText)) !== null) {
    imageIds.push(match[1]);
}

// Remove tags from text
finalText = finalText.replace(imageRegex, '')
                     .replace(/<[^>]+>/g, '')
                     .trim();

// Fetch image URLs from database
const imageUrls = [];
for (const pid of imageIds) {
    const product = await prisma.product.findUnique({ where: { id: pid.trim() } });
    if (product && product.images && product.images.length > 0) {
        const rawUrl = product.images[0];
        const cleanUrl = rawUrl.startsWith('http') ? rawUrl : `${process.env.BASE_URL}${rawUrl}`;
        imageUrls.push(cleanUrl);
    }
}

// Send text message
if (finalText) {
    await axios.post(
        `https://graph.facebook.com/v21.0/${process.env.WHATSAPP_PHONE_ID}/messages`,
        {
            messaging_product: "whatsapp",
            to: from,
            text: { body: finalText }
        },
        { headers: { 'Authorization': `Bearer ${process.env.WHATSAPP_ACCESS_TOKEN}`, 'Content-Type': 'application/json' } }
    );
}

// Send images separately
for (const cleanUrl of imageUrls) {
    await axios.post(
        `https://graph.facebook.com/v21.0/${process.env.WHATSAPP_PHONE_ID}/messages`,
        {
            messaging_product: "whatsapp",
            to: from,
            type: "image",
            image: { link: cleanUrl }
        },
        { headers: { 'Authorization': `Bearer ${process.env.WHATSAPP_ACCESS_TOKEN}`, 'Content-Type': 'application/json' } }
    );
}

worker.on('completed', job => {
    console.log(`Job ${job.id} has completed!`);
});

worker.on('failed', (job, err) => {
    console.log(`Job ${job.id} has failed with ${err.message}`);
});

let io;

export const setIO = (ioInstance) => {
    io = ioInstance;
    console.log("🔌 Socket.IO instance injected into Queue Worker");
};

app.set('io', io);
setIO(io); // Inject IO into Queue Worker