emailPollWorker.js

Go to the documentation of this file.

// Email Poll Worker: Polls tenant email settings and pushes inbound emails to automation
const Redis = require("ioredis");
const { ImapFlow } = require("imapflow");
const { simpleParser } = require("mailparser");
 
const redisConfig = require('./config/redis');
const redis = new Redis(redisConfig);
const db = require('./services/db');
 
/**
 *
 */
async function getTenantEmailSettings() {
  // Query DB for all tenants with email automation enabled
  const res = await db.query("SELECT tenant_id, email_automation FROM tenants WHERE email_automation IS NOT NULL");
  return res.rows;
}
 
/**
 *
 * @param tenantId
 * @param settings
 */
async function pollTenantMailbox(tenantId, settings) {
  // Validate settings before attempting connection
  if (!settings || typeof settings !== 'object') {
    console.log(`[EmailPoll] Tenant ${tenantId}: No email settings configured, skipping`);
    return;
  }
  
  const { host, port, username, password } = settings;
  
  // Skip if essential settings are missing
  if (!host || !port || !username || !password) {
    console.log(`[EmailPoll] Tenant ${tenantId}: Incomplete email config (missing host/port/username/password), skipping`);
    return;
  }
  
  // Skip localhost/loopback addresses (common misconfiguration)
  if (host === 'localhost' || host === '127.0.0.1' || host === '::1') {
    console.log(`[EmailPoll] Tenant ${tenantId}: Invalid host (${host}), skipping localhost connections`);
    return;
  }
  
  const client = new ImapFlow({
    host: settings.host,
    port: settings.port,
    secure: settings.security === "ssl",
    auth: {
      user: settings.username,
      pass: settings.password
    },
    logger: false // Suppress connection errors in logs
  });
  try {
    await client.connect();
    let lock = await client.getMailboxLock(settings.mailbox || "INBOX");
    try {
      // Search for unseen messages
      const messages = [];
      for await (let msg of client.fetch({ seen: false }, { source: true })) {
        const parsed = await simpleParser(msg.source);
        messages.push(parsed);
        redis.publish("emails:inbound", JSON.stringify({
          tenant_id: tenantId,
          from: parsed.from?.text,
          to: parsed.to?.text,
          subject: parsed.subject,
          text: parsed.text,
          html: parsed.html
        }));
      }
      if (messages.length > 0) {
        console.log(`[EmailPoll] Tenant ${tenantId}: Processed ${messages.length} new email(s)`);
      }
    } finally {
      lock.release();
    }
    await client.logout();
  } catch (err) {
    // Only log if it's not a connection refused error (common for unconfigured)
    if (err.code !== 'ECONNREFUSED' && err.code !== 'ETIMEDOUT') {
      console.error(`[EmailPoll] Tenant ${tenantId}: Error polling mailbox - ${err.message}`);
    }
  }
}
 
/**
 *
 */
async function pollAllTenants() {
  const tenants = await getTenantEmailSettings();
  for (const t of tenants) {
    try {
      await pollTenantMailbox(t.tenant_id, t.email_automation);
    } catch (e) { /* log error */ }
  }
}
 
setInterval(pollAllTenants, 60 * 1000); // Poll every minute
console.log("Email Poll Worker running: polling tenant mailboxes for inbound automation.");