132 lines
3.8 KiB
JavaScript
132 lines
3.8 KiB
JavaScript
#!/usr/bin/env node
|
|
'use strict';
|
|
|
|
/**
|
|
* Script to requeue messages from dead letter queue back to main queue
|
|
* Usage: node scripts/requeue_dlq_messages.js [--env <path>] [--dry-run] [--limit=10]
|
|
*
|
|
* Options:
|
|
* --env <path> Path to environment file (default: ./environment.env)
|
|
* --dry-run Show what would be done without doing it
|
|
* --limit=<num> Maximum messages to process (default: 10)
|
|
*/
|
|
|
|
const path = require('path');
|
|
|
|
// Parse command line arguments
|
|
const args = process.argv.slice(2);
|
|
let envFile = './environment.env';
|
|
|
|
for (let i = 0; i < args.length; i++) {
|
|
if (args[i] === '--env' && args[i + 1]) {
|
|
envFile = args[i + 1];
|
|
i++;
|
|
}
|
|
}
|
|
|
|
// Load environment variables
|
|
const envPath = path.resolve(process.cwd(), envFile);
|
|
console.log(`Loading environment from: ${envPath}`);
|
|
require('dotenv').config({ path: envPath });
|
|
|
|
const amqp = require('amqplib');
|
|
const env = require('../helpers/env');
|
|
|
|
const PARTNER_QUEUE = env.QUEUE_NAME_PARTNER;
|
|
const FAILED_QUEUE = `${PARTNER_QUEUE}_failed`;
|
|
|
|
async function requeueFromDLQ() {
|
|
const args = process.argv.slice(2);
|
|
const isDryRun = args.includes('--dry-run');
|
|
const limitArg = args.find(arg => arg.startsWith('--limit='));
|
|
const limit = limitArg ? parseInt(limitArg.split('=')[1]) : 10;
|
|
|
|
console.log(`Requeuing messages from failed queue: ${FAILED_QUEUE}`);
|
|
console.log(`Dry run: ${isDryRun}`);
|
|
console.log(`Limit: ${limit}`);
|
|
|
|
const conOps = {
|
|
protocol: 'amqp',
|
|
hostname: env.QUEUE_HOST || 'localhost',
|
|
port: env.QUEUE_PORT || 5672,
|
|
username: env.QUEUE_USR || 'agmuser',
|
|
password: env.QUEUE_PWD,
|
|
vhost: env.QUEUE_VHOST || '/',
|
|
heartbeat: env.QUEUE_HEARTBEAT || 0,
|
|
frameMax: 0
|
|
};
|
|
|
|
try {
|
|
const conn = await amqp.connect(conOps);
|
|
const ch = await conn.createChannel();
|
|
|
|
// Check failed queue status
|
|
const failedInfo = await ch.checkQueue(FAILED_QUEUE);
|
|
console.log(`\nFailed queue has ${failedInfo.messageCount} messages`);
|
|
|
|
if (failedInfo.messageCount === 0) {
|
|
console.log('No messages to requeue');
|
|
await conn.close();
|
|
return;
|
|
}
|
|
|
|
let requeuedCount = 0;
|
|
const maxToRequeue = Math.min(limit, failedInfo.messageCount);
|
|
|
|
console.log(`\nProcessing up to ${maxToRequeue} messages...`);
|
|
|
|
for (let i = 0; i < maxToRequeue; i++) {
|
|
const msg = await ch.get(FAILED_QUEUE);
|
|
|
|
if (!msg) {
|
|
console.log('No more messages in failed queue');
|
|
break;
|
|
}
|
|
|
|
try {
|
|
const taskMsg = JSON.parse(msg.content.toString());
|
|
console.log(`\nMessage ${i + 1}:`);
|
|
console.log(` Type: ${taskMsg.type}`);
|
|
console.log(` Data: ${JSON.stringify(taskMsg.data).substring(0, 100)}...`);
|
|
|
|
if (!isDryRun) {
|
|
// Republish to main queue
|
|
await ch.sendToQueue(PARTNER_QUEUE, msg.content, {
|
|
persistent: true
|
|
});
|
|
|
|
// Acknowledge the DLQ message to remove it
|
|
ch.ack(msg);
|
|
requeuedCount++;
|
|
console.log(` ✓ Requeued successfully`);
|
|
} else {
|
|
console.log(` ✓ Would be requeued (dry run)`);
|
|
// In dry run, we still need to ack to not block the queue
|
|
ch.ack(msg);
|
|
}
|
|
} catch (error) {
|
|
console.error(` ✗ Error processing message: ${error.message}`);
|
|
// Reject back to failed queue
|
|
ch.reject(msg, false);
|
|
}
|
|
}
|
|
|
|
console.log(`\nSummary:`);
|
|
console.log(`Messages processed: ${maxToRequeue}`);
|
|
console.log(`Messages requeued: ${isDryRun ? 0 : requeuedCount}`);
|
|
|
|
await conn.close();
|
|
|
|
} catch (error) {
|
|
console.error('Error:', error.message);
|
|
process.exit(1);
|
|
}
|
|
}
|
|
|
|
// Handle command line execution
|
|
if (require.main === module) {
|
|
requeueFromDLQ().catch(console.error);
|
|
}
|
|
|
|
module.exports = { requeueFromDLQ };
|