dispatch.ts

import { RabbitMQ } from "@snapstack/rabbitmq";
import fs from "fs";
import path from "path";
import prompts from "prompts";
import chalk from "chalk";

const MESSAGES_BASE_PATH = "./messages";
const OUTPUT_BASE_PATH = "./output";

// RabbitMQ Management API Configuration
const MANAGEMENT_API_URL = process.env.RABBITMQ_MANAGEMENT_URL || "http://localhost:15672";
const MANAGEMENT_USERNAME = process.env.RABBITMQ_MANAGEMENT_USER || "guest";
const MANAGEMENT_PASSWORD = process.env.RABBITMQ_MANAGEMENT_PASSWORD || "guest";

const rabbit = new RabbitMQ({
  url: "amqp://localhost:5672",
  reconnectTimeoutMs: 5000,
  prefetch: 10,
});

async function getAvailableFolders(): Promise<string[]> {
  try {
    const items = fs.readdirSync(MESSAGES_BASE_PATH);
    const folders = items.filter(item => {
      const fullPath = path.join(MESSAGES_BASE_PATH, item);
      return fs.statSync(fullPath).isDirectory();
    });
    return folders;
  } catch (error) {
    console.error(chalk.red("Error reading messages directory:"), error);
    return [];
  }
}

async function getAvailableQueues(): Promise<string[]> {
  try {
    console.log(chalk.blue("Fetching queues from RabbitMQ management API..."));
    
    // RabbitMQ management API endpoint
    const managementUrl = `${MANAGEMENT_API_URL}/api/queues`;
    
    // Use configured credentials
    const auth = Buffer.from(`${MANAGEMENT_USERNAME}:${MANAGEMENT_PASSWORD}`).toString('base64');
    
    const response = await fetch(managementUrl, {
      method: 'GET',
      headers: {
        'Authorization': `Basic ${auth}`,
        'Content-Type': 'application/json'
      }
    });
    
    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${response.statusText}`);
    }
    
    const queues = await response.json() as Array<{ name: string }>;
    
    // Extract queue names from the API response
    const queueNames = queues
      .map((queue) => queue.name)
      .filter((name: string) => name && name.trim() !== '') // Filter out empty names
      .sort(); // Sort alphabetically
    
    console.log(chalk.green(`✓ Found ${queueNames.length} queue(s) from management API`));
    
    if (queueNames.length > 0) {
      console.log(chalk.gray(`Queues: ${queueNames.slice(0, 5).join(', ')}${queueNames.length > 5 ? '...' : ''}`));
    }
    
    return queueNames;
    
  } catch (error) {
    console.log(chalk.yellow("⚠️ Could not fetch queues from management API. Using fallback list."));
    console.log(chalk.gray(`Error: ${error instanceof Error ? error.message : 'Unknown error'}`));
    
    // Fallback to common queue names if API is not available
    const fallbackQueues = [
      "shopify.events.queue",
      "shopify.inventory.queue",
      "orders.queue",
      "products.queue",
      "customers.queue"
    ];
    
    console.log(chalk.blue(`Using fallback list with ${fallbackQueues.length} common queue names`));
    return fallbackQueues;
  }
}

async function dispatchMessages(folderPath: string, queueName: string): Promise<number> {
  const fullFolderPath = path.join(MESSAGES_BASE_PATH, folderPath);
  
  let messageFiles;
  try {
    messageFiles = fs
      .readdirSync(fullFolderPath)
      .filter((file) => file.endsWith(".json"));
    console.log(chalk.blue(`Found ${messageFiles.length} messages in ${folderPath}`));
  } catch (error) {
    console.error(chalk.red("Error reading messages:"), error);
    return 0;
  }

  if (messageFiles.length === 0) {
    console.log(chalk.yellow("No JSON message files found in the selected folder."));
    return 0;
  }

  let count = 0;
  let messageContents = [];
  const batchSize = 100;

  console.log(chalk.blue(`\nStarting dispatch to queue: ${queueName}`));

  for (const message of messageFiles) {
    try {
      const messageContent = JSON.parse(
        fs.readFileSync(path.join(fullFolderPath, message), "utf8")
      );
      messageContents.push(messageContent);
      count++;

      if (count % batchSize === 0) {
        for (const content of messageContents) {
          await rabbit.publishToQueue(queueName, content);
        }
        console.log(chalk.green(`✓ Published batch of ${messageContents.length} messages (${count}/${messageFiles.length})`));
        messageContents = [];
      }
    } catch (error) {
      console.error(chalk.red(`Error processing message ${message}:`), error);
    }
  }

  // Publish any remaining messages in the final batch
  if (messageContents.length > 0) {
    for (const content of messageContents) {
      await rabbit.publishToQueue(queueName, content);
    }
    console.log(chalk.green(`✓ Published final batch of ${messageContents.length} messages`));
  }

  return count;
}

async function consumeMessages(queueName: string, outputFolder: string): Promise<void> {
  const fullOutputPath = path.join(OUTPUT_BASE_PATH, outputFolder);
  
  // Create output directory if it doesn't exist
  if (!fs.existsSync(OUTPUT_BASE_PATH)) {
    fs.mkdirSync(OUTPUT_BASE_PATH, { recursive: true });
  }
  
  if (!fs.existsSync(fullOutputPath)) {
    fs.mkdirSync(fullOutputPath, { recursive: true });
  }

  console.log(chalk.blue(`\nStarting consumption from queue: ${queueName}`));
  console.log(chalk.blue(`Saving messages to: ${fullOutputPath}`));
  console.log(chalk.yellow("Press Ctrl+C to stop consuming..."));

  let fileCount = 0;

  await rabbit.consume<any>(queueName, async (msg, raw) => {
    const { type } = msg;

    try {
      const fileName = `${type || 'message'}-${fileCount}.json`;
      const filePath = path.join(fullOutputPath, fileName);
      
      fs.writeFileSync(filePath, JSON.stringify(msg, null, 2));
      
      fileCount++;
      console.log(chalk.green(`✓ Saved message ${fileCount}: ${fileName}`));
    } catch (err) {
      console.error(chalk.red(`[RabbitMQ] Failed to process ${type || 'unknown'}:`), err);
      throw err; // Will cause the message to be NACKed
    }
  });
}

async function purgeQueue(queueName: string): Promise<boolean> {
  try {
    console.log(chalk.blue(`Purging queue "${queueName}"...`));
    
    let messageCount = 0;
    
    // Note: This is a simple approach that consumes messages to "purge" them
    // In a production environment, you might want to use RabbitMQ management API
    console.log(chalk.yellow(`Note: Consuming all messages from "${queueName}" to purge...`));
    
    // Create a promise that resolves when no more messages are received
    const purgePromise = new Promise<void>((resolve) => {
      let noMessageTimeout: NodeJS.Timeout;
      
      const resetTimeout = () => {
        if (noMessageTimeout) clearTimeout(noMessageTimeout);
        noMessageTimeout = setTimeout(() => {
          resolve(); // No messages received for 2 seconds, assume queue is empty
        }, 2000);
      };
      
      resetTimeout();
      
      rabbit.consume<any>(queueName, async (msg, raw) => {
        messageCount++;
        resetTimeout(); // Reset timeout since we got a message
        // Message is automatically acknowledged and removed
      });
    });
    
    // Wait for purging to complete or timeout after 30 seconds
    await Promise.race([
      purgePromise,
      new Promise(resolve => setTimeout(resolve, 30000))
    ]);
    
    console.log(chalk.green(`✓ Purged queue "${queueName}" - ${messageCount} messages removed`));
    return true;
  } catch (error) {
    console.error(chalk.red(`✗ Failed to purge queue "${queueName}":`), error);
    return false;
  }
}

async function handleDispatchFlow(): Promise<void> {
  // Get available folders
  console.log(chalk.blue("Scanning for message folders..."));
  const folders = await getAvailableFolders();
  
  if (folders.length === 0) {
    console.log(chalk.red("No folders found in ./messages directory"));
    return;
  }

  console.log(chalk.green(`✓ Found ${folders.length} folder(s)`));

  // Get available queues
  console.log(chalk.blue("Getting available queues..."));
  const queues = await getAvailableQueues();
  console.log(chalk.green(`✓ Found ${queues.length} queue(s)`));

  // Prompt user for folder selection
  const folderResponse = await prompts({
    type: 'select',
    name: 'folder',
    message: 'Which folder contains your messages?',
    choices: folders.map(folder => ({
      title: folder,
      description: `Messages in ./messages/${folder}`,
      value: folder
    }))
  });

  if (!folderResponse.folder) {
    console.log(chalk.yellow("No folder selected. Exiting."));
    return;
  }

  // Prompt user for queue selection
  const queueResponse = await prompts({
    type: 'select',
    name: 'queue',
    message: 'Which queue should receive the messages?',
    choices: [
      ...queues.map(queue => ({
        title: queue,
        description: `Existing queue: ${queue}`,
        value: queue
      })),
      {
        title: 'Custom queue name',
        description: 'Enter a custom queue name',
        value: '__custom__'
      }
    ]
  });

  if (!queueResponse.queue) {
    console.log(chalk.yellow("No queue selected. Exiting."));
    return;
  }

  let selectedQueue = queueResponse.queue;

  // Handle custom queue name
  if (selectedQueue === '__custom__') {
    const customQueueResponse = await prompts({
      type: 'text',
      name: 'customQueue',
      message: 'Enter custom queue name:',
      validate: value => value.length > 0 ? true : 'Queue name cannot be empty'
    });

    if (!customQueueResponse.customQueue) {
      console.log(chalk.yellow("No custom queue name provided. Exiting."));
      return;
    }

    selectedQueue = customQueueResponse.customQueue;
  }

  // Confirm before dispatching
  const confirmResponse = await prompts({
    type: 'confirm',
    name: 'confirm',
    message: `Dispatch messages from "${folderResponse.folder}" to "${selectedQueue}"?`,
    initial: true
  });

  if (!confirmResponse.confirm) {
    console.log(chalk.yellow("Operation cancelled."));
    return;
  }

  // Dispatch messages
  console.log(chalk.blue("\n📤 Dispatching messages..."));
  const totalMessages = await dispatchMessages(folderResponse.folder, selectedQueue);
  
  if (totalMessages > 0) {
    console.log(chalk.green.bold(`\n🎉 Successfully dispatched ${totalMessages} messages to ${selectedQueue}`));
  } else {
    console.log(chalk.yellow("No messages were dispatched."));
  }
}

async function handleConsumeFlow(): Promise<void> {
  // Get available queues
  console.log(chalk.blue("Getting available queues..."));
  const queues = await getAvailableQueues();
  console.log(chalk.green(`✓ Found ${queues.length} queue(s)`));

  // Prompt user for queue selection
  const queueResponse = await prompts({
    type: 'select',
    name: 'queue',
    message: 'Which queue should you consume messages from?',
    choices: [
      ...queues.map(queue => ({
        title: queue,
        description: `Consume from: ${queue}`,
        value: queue
      })),
      {
        title: 'Custom queue name',
        description: 'Enter a custom queue name',
        value: '__custom__'
      }
    ]
  });

  if (!queueResponse.queue) {
    console.log(chalk.yellow("No queue selected. Exiting."));
    return;
  }

  let selectedQueue = queueResponse.queue;

  // Handle custom queue name
  if (selectedQueue === '__custom__') {
    const customQueueResponse = await prompts({
      type: 'text',
      name: 'customQueue',
      message: 'Enter custom queue name:',
      validate: value => value.length > 0 ? true : 'Queue name cannot be empty'
    });

    if (!customQueueResponse.customQueue) {
      console.log(chalk.yellow("No custom queue name provided. Exiting."));
      return;
    }

    selectedQueue = customQueueResponse.customQueue;
  }

  // Prompt for output folder name
  const outputFolderResponse = await prompts({
    type: 'text',
    name: 'outputFolder',
    message: 'Enter output folder name:',
    initial: selectedQueue.replace(/\./g, '-'),
    validate: value => value.length > 0 ? true : 'Folder name cannot be empty'
  });

  if (!outputFolderResponse.outputFolder) {
    console.log(chalk.yellow("No output folder specified. Exiting."));
    return;
  }

  // Confirm before consuming
  const confirmResponse = await prompts({
    type: 'confirm',
    name: 'confirm',
    message: `Consume messages from "${selectedQueue}" to "./output/${outputFolderResponse.outputFolder}"?`,
    initial: true
  });

  if (!confirmResponse.confirm) {
    console.log(chalk.yellow("Operation cancelled."));
    return;
  }

  // Start consuming messages
  console.log(chalk.blue("\n📥 Starting message consumption..."));
  await consumeMessages(selectedQueue, outputFolderResponse.outputFolder);
}

async function handlePurgeFlow(): Promise<void> {
  // Get available queues
  console.log(chalk.blue("Getting available queues..."));
  const queues = await getAvailableQueues();
  console.log(chalk.green(`✓ Found ${queues.length} queue(s)`