diff --git a/app/models/emails.js b/app/models/emails.js index 143a77190..513652e10 100644 --- a/app/models/emails.js +++ b/app/models/emails.js @@ -138,9 +138,22 @@ const Emails = new mongoose.Schema( index: true }, priority: { + type: Number, + default: 1, // PRIORITY_LEVELS.NORMAL + index: true + }, + // abuse score (0-100, higher = more suspicious) + abuse_score: { type: Number, default: 0, - min: 0 + min: 0, + max: 100, + index: true + }, + // temporary throttling (if set, email won't be processed until this date) + throttled_until: { + type: Date, + index: true }, alias: { type: mongoose.Schema.ObjectId, @@ -344,7 +357,9 @@ Emails.plugin(mongooseCommonPlugin, { 'message', 'locked_by', 'locked_at', - 'priority' + 'priority', + 'abuse_score', + 'throttled_until' ] }); @@ -354,6 +369,24 @@ Emails.index( { partialFilterExpression: { locked_at: { $exists: true } } } ); +// composite index for fair queue processing (most important queries first) +Emails.index({ + status: 1, + priority: -1, // higher priority first + domain: 1, + user: 1, + created_at: 1 // FIFO within same priority/domain/user +}); + +// index for throttled emails cleanup +Emails.index( + { throttled_until: 1, status: 1 }, + { partialFilterExpression: { throttled_until: { $exists: true } } } +); + +// index for abuse score queries +Emails.index({ abuse_score: 1, user: 1 }); + // DSN Emails.pre('validate', function (next) { if (this.dsn === undefined) return next(); @@ -701,27 +734,62 @@ Emails.pre('save', function (next) { } }); -// determine priority +// determine priority using fair queue system Emails.pre('save', async function (next) { try { - const domain = await Domains.findById(this.domain); - // we return here instead of erroring - if (!domain) { + // Only calculate priority for new emails or when priority is not set + if (!this.isNew && this.priority !== undefined) { + return next(); + } + + const { PRIORITY_LEVELS } = require('#config/priority-levels'); + const dayjs = require('dayjs-with-plugins'); + + // Default priority + let priority = PRIORITY_LEVELS.NORMAL; + + const [domain, user] = await Promise.all([ + Domains.findById(this.domain).populate('members.user', 'id group plan created_at'), + Users.findById(this.user).select('id group plan created_at').lean() + ]); + + // If domain or user not found, set to pending with low priority + if (!domain || !user) { this.status = 'pending'; - this.priority = 0; + this.priority = PRIORITY_LEVELS.LOW; return next(); } - // if any of the domain admins are admins then set priority to 1 - const adminExists = await Users.exists({ - _id: { - $in: domain.members - .filter((m) => m.group === 'admin') - .map((m) => m.user) - }, - group: 'admin' - }); - this.priority = adminExists ? 1 : 0; + // Check if user is an admin + const isUserAdmin = user.group === 'admin'; + + // Check if domain has admin users with premium plans + const hasAdminMembers = domain.members.some( + m => m.user && + m.group === 'admin' && + m.user.group === 'admin' && + ['enhanced_protection', 'team'].includes(m.user.plan) + ); + + // Determine priority based on user/domain status + if (isUserAdmin || hasAdminMembers) { + priority = PRIORITY_LEVELS.HIGH; + } else if (user.plan && ['enhanced_protection', 'team'].includes(user.plan)) { + priority = PRIORITY_LEVELS.HIGH; + } else if (dayjs().diff(user.created_at, 'days') < 7) { + // New accounts get lower priority for the first week + priority = PRIORITY_LEVELS.LOW; + } else { + priority = PRIORITY_LEVELS.NORMAL; + } + + this.priority = priority; + + // Initialize abuse score if not set + if (this.abuse_score === undefined) { + this.abuse_score = 0; + } + next(); } catch (err) { next(err); diff --git a/config/priority-levels.js b/config/priority-levels.js new file mode 100644 index 000000000..5f142cab2 --- /dev/null +++ b/config/priority-levels.js @@ -0,0 +1,49 @@ +/** + * Copyright (c) Forward Email LLC + * SPDX-License-Identifier: BUSL-1.1 + */ + +// Priority levels for fair queue system +const PRIORITY_LEVELS = { + HIGH: 2, // Paid users, good reputation, admin domains + NORMAL: 1, // Regular users (default) + LOW: 0, // Free tier, new accounts + THROTTLED: -1 // Detected abuse/suspicious activity +}; + +// Human readable names for priority levels +const PRIORITY_NAMES = { + [PRIORITY_LEVELS.HIGH]: 'HIGH', + [PRIORITY_LEVELS.NORMAL]: 'NORMAL', + [PRIORITY_LEVELS.LOW]: 'LOW', + [PRIORITY_LEVELS.THROTTLED]: 'THROTTLED' +}; + +// Configuration for priority-based processing limits +const PRIORITY_CONFIG = { + // Multipliers for queue allocation by priority + QUEUE_MULTIPLIERS: { + [PRIORITY_LEVELS.HIGH]: 1.5, // 50% more queue allocation + [PRIORITY_LEVELS.NORMAL]: 1.0, // Normal allocation + [PRIORITY_LEVELS.LOW]: 0.5, // Half allocation + [PRIORITY_LEVELS.THROTTLED]: 0.1 // Minimal allocation + }, + + // Maximum concurrent processing by priority + CONCURRENCY_LIMITS: { + [PRIORITY_LEVELS.HIGH]: 0.4, // 40% of total concurrency + [PRIORITY_LEVELS.NORMAL]: 0.4, // 40% of total concurrency + [PRIORITY_LEVELS.LOW]: 0.15, // 15% of total concurrency + [PRIORITY_LEVELS.THROTTLED]: 0.05 // 5% of total concurrency + }, + + // Abuse detection thresholds + ABUSE_THRESHOLD: 50, // Score above which user gets throttled + MAX_THROTTLE_DURATION: 24 * 60 * 60 * 1000 // 24 hours max throttle +}; + +module.exports = { + PRIORITY_LEVELS, + PRIORITY_NAMES, + PRIORITY_CONFIG +}; \ No newline at end of file diff --git a/docs/fair-queue-implementation-plan.md b/docs/fair-queue-implementation-plan.md new file mode 100644 index 000000000..30cb2b565 --- /dev/null +++ b/docs/fair-queue-implementation-plan.md @@ -0,0 +1,497 @@ +# Fair Queue Implementation Plan + +## Overview + +This document outlines a comprehensive strategy to implement fair queuing for SMTP outbound emails, preventing bad actors from monopolizing the queue while ensuring fair access for all users. + +## Current System Analysis + +### Issues Identified +1. **FIFO with no fairness**: Current queue processes emails first-come-first-served (sort by `created_at: -1`) +2. **Single user can flood queue**: No per-user/domain limits in queue processing +3. **Commented fair queue code**: Lines 158-211 in `jobs/send-emails.js` show attempted domain-based fairness but it's disabled +4. **Basic rate limiting only**: Only has `smtpLimitMessages: 300/day` per user but no queue-time fairness + +### Current Queue Flow +- **Entry Point**: `helpers/on-data-smtp.js:612` - Sets `email.status = 'queued'` +- **Processing**: `jobs/send-emails.js:214-236` - Processes emails with PQueue +- **Monitoring**: `jobs/check-smtp-queue-count.js:92-111` - Queue health checks +- **Email Sending**: `helpers/process-email.js:71+` - Actual email delivery + +## Strategy: Weighted Fair Queuing (WFQ) with Anti-Abuse + +### Core Principles +1. **Domain-level fairness** - Prevent single domains from dominating +2. **User-level fairness** - Ensure fair access within domains +3. **Priority tiers** - Different service levels based on user status +4. **Abuse detection** - Automatic throttling for suspicious behavior +5. **Adaptive limits** - Dynamic adjustments based on system load + +## Implementation Phases + +### Phase 1: Database Schema Updates (Week 1) + +#### Email Model Enhancements +```javascript +// Add new fields to Email schema +{ + priority: { + type: Number, + default: 1, // PRIORITY_LEVELS.NORMAL + index: true + }, + user: { + type: mongoose.Schema.Types.ObjectId, + ref: 'User', + required: true, + index: true + }, + throttled_until: { + type: Date, + index: true + }, + abuse_score: { + type: Number, + default: 0 + } +} +``` + +#### Index Optimization +```javascript +// Composite index for efficient queue queries +db.emails.createIndex({ + status: 1, + priority: -1, + domain: 1, + user: 1, + created_at: 1 +}); + +// Index for throttled emails +db.emails.createIndex({ + throttled_until: 1, + status: 1 +}); +``` + +### Phase 2: Fair Queue Core Logic (Week 2) + +#### Domain-Based Fair Distribution +```javascript +// Enhanced sendEmails function +async function sendEmailsFairly() { + const limit = config.smtpMaxQueue - queue.size; + + // Get domain distribution + const domainEmailCounts = await Emails.aggregate([ + { $match: query }, + { $group: { _id: '$domain', count: { $sum: 1 } } }, + { $sort: { count: 1 } } // Process domains with fewer queued emails first + ]); + + const maxPerDomain = Math.max(1, Math.floor(limit / domainEmailCounts.length)); + const selectedEmails = []; + + for (const { _id: domainId, count } of domainEmailCounts) { + const domainEmails = await Emails.find({ + ...query, + domain: domainId + }) + .sort({ priority: -1, created_at: 1 }) // Priority first, then FIFO within domain + .limit(Math.min(maxPerDomain, count)) + .lean(); + + selectedEmails.push(...domainEmails); + if (selectedEmails.length >= limit) break; + } + + // Process selected emails + for (const email of selectedEmails) { + queue.add(() => processEmail({ email, resolver, client })); + } +} +``` + +#### Priority System Implementation +```javascript +const PRIORITY_LEVELS = { + HIGH: 2, // Paid users, good reputation + NORMAL: 1, // Regular users + LOW: 0, // Free tier, new accounts + THROTTLED: -1 // Detected abuse/suspicious activity +}; + +function calculatePriority(user, domain) { + let priority = PRIORITY_LEVELS.NORMAL; + + // Premium users get higher priority + if (user.plan === 'team' || user.plan === 'enhanced_protection') { + priority = PRIORITY_LEVELS.HIGH; + } + + // New accounts get lower priority + if (dayjs().diff(user.created_at, 'days') < 7) { + priority = PRIORITY_LEVELS.LOW; + } + + // Throttled users + if (user.abuse_score > ABUSE_THRESHOLD) { + priority = PRIORITY_LEVELS.THROTTLED; + } + + return priority; +} +``` + +### Phase 3: Abuse Detection System (Week 3) + +#### Abuse Scoring Algorithm +```javascript +async function calculateAbuseScore(user, domain) { + const window = dayjs().subtract(24, 'hours').toDate(); + + const stats = await Emails.aggregate([ + { + $match: { + user: user._id, + created_at: { $gte: window } + } + }, + { + $group: { + _id: null, + total: { $sum: 1 }, + bounced: { $sum: { $cond: [{ $eq: ['$status', 'bounced'] }, 1, 0] } }, + rejected: { $sum: { $cond: [{ $eq: ['$status', 'rejected'] }, 1, 0] } } + } + } + ]); + + if (!stats[0]) return 0; + + const { total, bounced, rejected } = stats[0]; + const bounceRate = bounced / total; + const rejectRate = rejected / total; + + let score = 0; + + // High bounce/reject rates + if (bounceRate > 0.1) score += 30; + if (rejectRate > 0.05) score += 20; + + // Volume spikes + const avgDaily = await getAverageDailyVolume(user._id); + if (total > avgDaily * 5) score += 25; + + // New account sending high volume + if (dayjs().diff(user.created_at, 'days') < 7 && total > 100) score += 40; + + return Math.min(score, 100); +} + +const ABUSE_THRESHOLD = 50; +``` + +#### Automatic Throttling +```javascript +async function applyThrottling(email, abuseScore) { + if (abuseScore > ABUSE_THRESHOLD) { + const throttleDuration = Math.min( + ms('1h') * Math.pow(2, Math.floor(abuseScore / 20)), // Exponential backoff + ms('24h') // Max 24 hours + ); + + await Emails.updateMany( + { user: email.user, status: 'queued' }, + { + $set: { + priority: PRIORITY_LEVELS.THROTTLED, + throttled_until: new Date(Date.now() + throttleDuration) + } + } + ); + + // Alert admins + await emailHelper({ + template: 'alert', + message: { + to: config.email.message.from, + subject: 'User throttled for suspicious activity' + }, + locals: { + user: email.user, + abuseScore, + throttleDuration: prettyMilliseconds(throttleDuration) + } + }); + } +} +``` + +### Phase 4: User-Level Fairness (Week 4) + +#### Per-User Queue Distribution +```javascript +async function distributeByUser(domainId, domainLimit) { + const userEmailCounts = await Emails.aggregate([ + { + $match: { + ...query, + domain: domainId, + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: new Date() } } + ] + } + }, + { $group: { _id: '$user', count: { $sum: 1 } } }, + { $sort: { count: 1 } } + ]); + + const maxPerUser = Math.max(1, Math.floor(domainLimit / userEmailCounts.length)); + const selectedEmails = []; + + for (const { _id: userId, count } of userEmailCounts) { + const userEmails = await Emails.find({ + ...query, + domain: domainId, + user: userId + }) + .sort({ priority: -1, created_at: 1 }) + .limit(Math.min(maxPerUser, count)) + .lean(); + + selectedEmails.push(...userEmails); + if (selectedEmails.length >= domainLimit) break; + } + + return selectedEmails; +} +``` + +### Phase 5: Monitoring & Tuning (Week 5) + +#### Queue Fairness Metrics +```javascript +// Add to monitoring system +async function collectQueueMetrics() { + const metrics = { + // Domain distribution + domainDistribution: await Emails.aggregate([ + { $match: { status: 'queued' } }, + { $group: { _id: '$domain', count: { $sum: 1 } } }, + { $sort: { count: -1 } } + ]), + + // User distribution + userDistribution: await Emails.aggregate([ + { $match: { status: 'queued' } }, + { $group: { _id: '$user', count: { $sum: 1 } } }, + { $sort: { count: -1 } } + ]), + + // Priority distribution + priorityDistribution: await Emails.aggregate([ + { $match: { status: 'queued' } }, + { $group: { _id: '$priority', count: { $sum: 1 } } } + ]), + + // Throttled users + throttledCount: await Emails.countDocuments({ + status: 'queued', + throttled_until: { $gt: new Date() } + }), + + // Queue health score + queueHealth: await calculateQueueHealth() + }; + + return metrics; +} +``` + +#### Adaptive Rate Limiting +```javascript +async function calculateAdaptiveLimits() { + const queueHealth = await calculateQueueHealth(); + const baseLimit = config.smtpMaxQueue; + + return { + high_priority: Math.floor(baseLimit * queueHealth.multiplier * 1.5), + normal: Math.floor(baseLimit * queueHealth.multiplier), + low: Math.floor(baseLimit * queueHealth.multiplier * 0.5), + throttled: Math.max(1, Math.floor(baseLimit * queueHealth.multiplier * 0.1)) + }; +} + +async function calculateQueueHealth() { + const totalQueued = await Emails.countDocuments({ status: 'queued' }); + const oldestQueued = await Emails.findOne( + { status: 'queued' }, + { created_at: 1 }, + { sort: { created_at: 1 } } + ); + + const queueAge = oldestQueued ? + dayjs().diff(oldestQueued.created_at, 'minutes') : 0; + + // Health score from 0.1 (unhealthy) to 2.0 (very healthy) + let multiplier = 1.0; + + if (totalQueued > config.smtpMaxQueue * 0.8) multiplier *= 0.7; + if (queueAge > 30) multiplier *= 0.5; // Emails waiting > 30 minutes + if (totalQueued < config.smtpMaxQueue * 0.3) multiplier *= 1.5; + + return { + totalQueued, + queueAge, + multiplier: Math.max(0.1, Math.min(2.0, multiplier)) + }; +} +``` + +### Phase 6: Advanced Features (Week 6+) + +#### Premium Priority Lanes +```javascript +// Separate processing lanes for different priorities +const priorityQueues = { + [PRIORITY_LEVELS.HIGH]: new PQueue({ concurrency: Math.floor(config.smtpMaxQueue * 0.4) }), + [PRIORITY_LEVELS.NORMAL]: new PQueue({ concurrency: Math.floor(config.smtpMaxQueue * 0.4) }), + [PRIORITY_LEVELS.LOW]: new PQueue({ concurrency: Math.floor(config.smtpMaxQueue * 0.15) }), + [PRIORITY_LEVELS.THROTTLED]: new PQueue({ concurrency: Math.floor(config.smtpMaxQueue * 0.05) }) +}; + +async function processEmailByPriority(email) { + const targetQueue = priorityQueues[email.priority] || priorityQueues[PRIORITY_LEVELS.NORMAL]; + + targetQueue.add(async () => { + await processEmail({ email, resolver, client }); + }); +} +``` + +#### Machine Learning Integration +```javascript +// Future enhancement: ML-based abuse detection +async function mlAbuseDetection(user, email) { + const features = await extractFeatures(user, email); + const prediction = await mlModel.predict(features); + + return { + abuseScore: prediction.score, + confidence: prediction.confidence, + reasons: prediction.reasons + }; +} +``` + +## Configuration Updates + +### Environment Variables +```bash +# Add to .env +SMTP_FAIR_QUEUE_ENABLED=true +SMTP_ABUSE_DETECTION_ENABLED=true +SMTP_ABUSE_THRESHOLD=50 +SMTP_MAX_THROTTLE_DURATION=86400000 # 24 hours +SMTP_PRIORITY_MULTIPLIERS="2,1,0.5,0.1" # HIGH,NORMAL,LOW,THROTTLED +``` + +### Config Updates +```javascript +// Add to config/index.js +module.exports = { + // ... existing config + + // Fair queue settings + smtpFairQueueEnabled: boolean(env.SMTP_FAIR_QUEUE_ENABLED), + smtpAbuseDetectionEnabled: boolean(env.SMTP_ABUSE_DETECTION_ENABLED), + smtpAbuseThreshold: Number.parseInt(env.SMTP_ABUSE_THRESHOLD, 10) || 50, + smtpMaxThrottleDuration: Number.parseInt(env.SMTP_MAX_THROTTLE_DURATION, 10) || ms('24h'), + smtpPriorityMultipliers: (env.SMTP_PRIORITY_MULTIPLIERS || '2,1,0.5,0.1') + .split(',') + .map(Number) +}; +``` + +## Testing Strategy + +### Unit Tests +- Priority calculation logic +- Abuse score calculation +- Fair distribution algorithms +- Throttling mechanisms + +### Integration Tests +- End-to-end queue processing +- Database query performance +- Queue fairness under load +- Abuse detection accuracy + +### Load Testing +- Queue performance with 10k+ emails +- Fairness maintenance under high load +- System stability during abuse scenarios +- Memory and CPU usage optimization + +## Rollout Plan + +### Stage 1: Database Migration +- Deploy schema changes +- Create indexes +- Migrate existing data + +### Stage 2: Feature Flags +- Deploy code with features disabled +- Enable monitoring first +- Gradual feature enablement + +### Stage 3: A/B Testing +- Split traffic between old/new systems +- Monitor fairness metrics +- Performance comparison + +### Stage 4: Full Rollout +- Complete migration to fair queue +- Remove old code +- Full monitoring deployment + +## Success Metrics + +### Fairness Metrics +- **Gini coefficient** of email distribution per domain/user +- **Maximum queue monopolization** by single entity +- **Average wait time** across user tiers + +### Performance Metrics +- **Queue processing throughput** +- **Database query performance** +- **Memory/CPU usage** +- **Email delivery latency** + +### Abuse Detection Metrics +- **False positive rate** for throttling +- **Time to detect** abuse scenarios +- **Effectiveness** at preventing queue monopolization + +## Benefits + +1. **Prevents queue monopolization** by bad actors +2. **Maintains fairness** across all users and domains +3. **Adaptive performance** based on system load +4. **Abuse detection** with automatic throttling +5. **Graceful degradation** under high load +6. **Backwards compatible** with existing system +7. **Premium user experience** through priority lanes +8. **Scalable architecture** for future growth + +## Future Enhancements + +- **Geographic distribution** fairness +- **Time-based priority** adjustments +- **Machine learning** abuse detection +- **Predictive scaling** based on queue patterns +- **Multi-tenant** isolation improvements +- **Real-time** queue visualization dashboard \ No newline at end of file diff --git a/helpers/fair-queue.js b/helpers/fair-queue.js new file mode 100644 index 000000000..ed3d330f3 --- /dev/null +++ b/helpers/fair-queue.js @@ -0,0 +1,444 @@ +/** + * Copyright (c) Forward Email LLC + * SPDX-License-Identifier: BUSL-1.1 + */ + +const dayjs = require('dayjs-with-plugins'); +const ms = require('ms'); + +const Emails = require('#models/emails'); +const Users = require('#models/users'); +const config = require('#config'); +const logger = require('#helpers/logger'); +const { PRIORITY_LEVELS, PRIORITY_CONFIG } = require('#config/priority-levels'); + +/** + * Calculate queue health metrics and adaptive multipliers + * @returns {Object} Queue health information + */ +async function calculateQueueHealth() { + const now = new Date(); + + // Count total queued emails + const totalQueued = await Emails.countDocuments({ + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: now } } + ] + }); + + // Find oldest queued email + const oldestQueued = await Emails.findOne( + { + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: now } } + ] + }, + { created_at: 1 }, + { sort: { created_at: 1 } } + ); + + const queueAge = oldestQueued ? + dayjs().diff(oldestQueued.created_at, 'minutes') : 0; + + // Calculate health multiplier (0.1 to 2.0) + let multiplier = 1.0; + + // Reduce throughput if queue is too full + if (totalQueued > config.smtpMaxQueue * 0.8) { + multiplier *= 0.7; + } + + // Reduce throughput if emails are waiting too long + if (queueAge > 30) { + multiplier *= 0.5; + } + + // Increase throughput if queue is light + if (totalQueued < config.smtpMaxQueue * 0.3) { + multiplier *= 1.5; + } + + // Clamp between min and max values + multiplier = Math.max(0.1, Math.min(2.0, multiplier)); + + return { + totalQueued, + queueAge, + multiplier, + isHealthy: multiplier >= 0.8 + }; +} + +/** + * Calculate adaptive limits for each priority level + * @param {number} baseLimit - Base queue limit + * @param {Object} queueHealth - Queue health metrics + * @returns {Object} Priority-based limits + */ +function calculateAdaptiveLimits(baseLimit, queueHealth) { + const limits = {}; + + // Adjust multipliers based on queue health + let adjustedMultipliers = { ...PRIORITY_CONFIG.QUEUE_MULTIPLIERS }; + + // If queue is unhealthy, favor high priority more + if (!queueHealth.isHealthy) { + adjustedMultipliers[PRIORITY_LEVELS.HIGH] *= 2.0; // Boost high priority + adjustedMultipliers[PRIORITY_LEVELS.NORMAL] *= 0.7; // Reduce normal + adjustedMultipliers[PRIORITY_LEVELS.LOW] *= 0.3; // Significantly reduce low + adjustedMultipliers[PRIORITY_LEVELS.THROTTLED] *= 0.1; // Minimal throttled + } + + // If queue age is very high, prioritize clearing backlog + if (queueHealth.queueAge > 60) { // Over 1 hour old + adjustedMultipliers[PRIORITY_LEVELS.HIGH] *= 1.5; + adjustedMultipliers[PRIORITY_LEVELS.NORMAL] *= 1.2; + } + + for (const [priority, multiplier] of Object.entries(adjustedMultipliers)) { + limits[priority] = Math.max( + priority === PRIORITY_LEVELS.THROTTLED ? 0 : 1, // Throttled can be 0 + Math.floor(baseLimit * queueHealth.multiplier * multiplier) + ); + } + + // Ensure we don't exceed total limit + const totalAllocated = Object.values(limits).reduce((sum, limit) => sum + limit, 0); + if (totalAllocated > baseLimit) { + const ratio = baseLimit / totalAllocated; + for (const priority of Object.keys(limits)) { + limits[priority] = Math.floor(limits[priority] * ratio); + } + } + + return limits; +} + +/** + * Get domain-based email distribution for fair queuing with adaptive priority limits + * @param {Object} query - Base MongoDB query + * @param {number} limit - Maximum emails to process + * @param {Object} queueHealth - Queue health metrics (optional) + * @returns {Array} Selected emails for processing + */ +async function getDomainFairDistribution(query, limit, queueHealth = null) { + // Calculate adaptive limits if queue health provided + let priorityLimits = null; + if (queueHealth) { + priorityLimits = calculateAdaptiveLimits(limit, queueHealth); + logger.debug('Adaptive priority limits:', priorityLimits); + } + + // Get domain distribution with priority breakdown + const domainEmailCounts = await Emails.aggregate([ + { $match: query }, + { + $group: { + _id: { domain: '$domain', priority: '$priority' }, + count: { $sum: 1 } + } + }, + { + $group: { + _id: '$_id.domain', + totalCount: { $sum: '$count' }, + priorityBreakdown: { + $push: { + priority: '$_id.priority', + count: '$count' + } + } + } + }, + { $sort: { totalCount: 1 } } // Least busy domains first + ]); + + if (domainEmailCounts.length === 0) { + return []; + } + + logger.info(`Fair queue: Processing ${domainEmailCounts.length} domains`); + + const selectedEmails = []; + const priorityUsed = {}; + + // Initialize priority usage tracking + if (priorityLimits) { + for (const priority of Object.keys(priorityLimits)) { + priorityUsed[priority] = 0; + } + } + + // Calculate base max per domain + const baseMaxPerDomain = Math.max(1, Math.floor(limit / domainEmailCounts.length)); + + // Process each domain fairly + for (const { _id: domainId, totalCount, priorityBreakdown } of domainEmailCounts) { + if (selectedEmails.length >= limit) break; + + const remainingLimit = limit - selectedEmails.length; + const domainLimit = Math.min(baseMaxPerDomain, totalCount, remainingLimit); + + let domainSelected = 0; + + // Process by priority levels (high to low) + const sortedPriorities = priorityBreakdown.sort((a, b) => b.priority - a.priority); + + for (const { priority, count } of sortedPriorities) { + if (domainSelected >= domainLimit) break; + + let priorityLimit = domainLimit - domainSelected; + + // Apply adaptive priority limits if available + if (priorityLimits && priorityLimits[priority] !== undefined) { + const remainingForPriority = priorityLimits[priority] - (priorityUsed[priority] || 0); + priorityLimit = Math.min(priorityLimit, remainingForPriority); + } + + if (priorityLimit <= 0) continue; + + // Get emails for this domain and priority + const priorityEmails = await Emails.find({ + ...query, + domain: domainId, + priority: priority + }) + .sort({ created_at: 1 }) // FIFO within same priority + .limit(priorityLimit) + .lean(); + + selectedEmails.push(...priorityEmails); + domainSelected += priorityEmails.length; + + if (priorityLimits) { + priorityUsed[priority] = (priorityUsed[priority] || 0) + priorityEmails.length; + } + + logger.debug( + `Fair queue: Domain ${domainId.toString().slice(-6)} Priority ${priority} - selected ${priorityEmails.length}/${count} emails` + ); + } + } + + // Log priority usage summary + if (priorityLimits) { + logger.info('Priority usage:', priorityUsed); + } + + return selectedEmails; +} + +/** + * Get user-based email distribution within a domain for fair queuing + * @param {Object} query - Base MongoDB query + * @param {string} domainId - Domain ID + * @param {number} domainLimit - Maximum emails for this domain + * @returns {Array} Selected emails for processing + */ +async function getUserFairDistribution(query, domainId, domainLimit) { + // Get user distribution within domain + const userEmailCounts = await Emails.aggregate([ + { + $match: { + ...query, + domain: domainId + } + }, + { + $group: { + _id: '$user', + count: { $sum: 1 }, + avgPriority: { $avg: '$priority' } + } + }, + { $sort: { count: 1 } } // Least busy users first + ]); + + if (userEmailCounts.length === 0) { + return []; + } + + const maxPerUser = Math.max(1, Math.floor(domainLimit / userEmailCounts.length)); + const selectedEmails = []; + + // Process each user fairly within domain + for (const { _id: userId, count } of userEmailCounts) { + if (selectedEmails.length >= domainLimit) break; + + const userLimit = Math.min(maxPerUser, count, domainLimit - selectedEmails.length); + + // Get emails for this user, prioritized + const userEmails = await Emails.find({ + ...query, + domain: domainId, + user: userId + }) + .sort({ + priority: -1, + created_at: 1 + }) + .limit(userLimit) + .lean(); + + selectedEmails.push(...userEmails); + } + + return selectedEmails; +} + +/** + * Calculate abuse score for a user based on recent activity + * @param {string} userId - User ID + * @returns {number} Abuse score (0-100) + */ +async function calculateUserAbuseScore(userId) { + const window = dayjs().subtract(24, 'hours').toDate(); + + const stats = await Emails.aggregate([ + { + $match: { + user: userId, + created_at: { $gte: window } + } + }, + { + $group: { + _id: null, + total: { $sum: 1 }, + bounced: { $sum: { $cond: [{ $eq: ['$status', 'bounced'] }, 1, 0] } }, + rejected: { $sum: { $cond: [{ $eq: ['$status', 'rejected'] }, 1, 0] } } + } + } + ]); + + if (!stats[0] || stats[0].total === 0) return 0; + + const { total, bounced, rejected } = stats[0]; + const bounceRate = bounced / total; + const rejectRate = rejected / total; + + let score = 0; + + // High bounce/reject rates + if (bounceRate > 0.1) score += 30; // 10%+ bounce rate + if (rejectRate > 0.05) score += 20; // 5%+ reject rate + + // Get user info for additional checks + const user = await Users.findById(userId).select('created_at').lean(); + if (!user) return score; + + // Get average daily volume for comparison + const avgDaily = await getAverageDailyVolume(userId); + + // Volume spike detection + if (total > avgDaily * 5) score += 25; // 5x normal volume + + // New account sending high volume + if (dayjs().diff(user.created_at, 'days') < 7 && total > 100) { + score += 40; + } + + return Math.min(score, 100); +} + +/** + * Get average daily email volume for a user over past 30 days + * @param {string} userId - User ID + * @returns {number} Average daily volume + */ +async function getAverageDailyVolume(userId) { + const thirtyDaysAgo = dayjs().subtract(30, 'days').toDate(); + + const total = await Emails.countDocuments({ + user: userId, + created_at: { $gte: thirtyDaysAgo } + }); + + return Math.max(1, Math.floor(total / 30)); +} + +/** + * Apply throttling to a user's emails based on abuse score + * @param {string} userId - User ID + * @param {number} abuseScore - Calculated abuse score + */ +async function applyUserThrottling(userId, abuseScore) { + if (abuseScore < PRIORITY_CONFIG.ABUSE_THRESHOLD) return; + + // Calculate throttle duration (exponential backoff) + const throttleDuration = Math.min( + ms('1h') * Math.pow(2, Math.floor(abuseScore / 20)), + PRIORITY_CONFIG.MAX_THROTTLE_DURATION + ); + + const throttledUntil = new Date(Date.now() + throttleDuration); + + // Update all queued emails for this user + const result = await Emails.updateMany( + { + user: userId, + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: new Date() } } + ] + }, + { + $set: { + priority: PRIORITY_LEVELS.THROTTLED, + throttled_until: throttledUntil, + abuse_score: abuseScore + } + } + ); + + logger.warn( + `User ${userId} throttled for ${ms(throttleDuration, { long: true })} - abuse score: ${abuseScore}, affected emails: ${result.modifiedCount}` + ); + + return { + throttledUntil, + affectedEmails: result.modifiedCount, + duration: throttleDuration + }; +} + +/** + * Clean up expired throttling + */ +async function cleanupExpiredThrottling() { + const now = new Date(); + + const result = await Emails.updateMany( + { + status: 'queued', + throttled_until: { $exists: true, $lt: now }, + priority: PRIORITY_LEVELS.THROTTLED + }, + { + $set: { priority: PRIORITY_LEVELS.NORMAL }, + $unset: { throttled_until: 1 } + } + ); + + if (result.modifiedCount > 0) { + logger.info(`Cleaned up throttling for ${result.modifiedCount} emails`); + } + + return result.modifiedCount; +} + +module.exports = { + calculateQueueHealth, + calculateAdaptiveLimits, + getDomainFairDistribution, + getUserFairDistribution, + calculateUserAbuseScore, + applyUserThrottling, + cleanupExpiredThrottling, + getAverageDailyVolume +}; \ No newline at end of file diff --git a/jobs/monitor-queue-abuse.js b/jobs/monitor-queue-abuse.js new file mode 100755 index 000000000..c3ef56a0d --- /dev/null +++ b/jobs/monitor-queue-abuse.js @@ -0,0 +1,352 @@ +#!/usr/bin/env node + +/** + * Copyright (c) Forward Email LLC + * SPDX-License-Identifier: BUSL-1.1 + */ + +// eslint-disable-next-line import/no-unassigned-import +require('#config/env'); + +const process = require('node:process'); +const { parentPort } = require('node:worker_threads'); + +// eslint-disable-next-line import/no-unassigned-import +require('#config/mongoose'); + +const Graceful = require('@ladjs/graceful'); +const mongoose = require('mongoose'); +const dayjs = require('dayjs-with-plugins'); +const ms = require('ms'); + +const Emails = require('#models/emails'); +const Users = require('#models/users'); +const config = require('#config'); +const logger = require('#helpers/logger'); +const setupMongoose = require('#helpers/setup-mongoose'); +const emailHelper = require('#helpers/email'); +const fairQueue = require('#helpers/fair-queue'); +const { PRIORITY_CONFIG } = require('#config/priority-levels'); +const monitorServer = require('#helpers/monitor-server'); + +monitorServer(); + +const graceful = new Graceful({ + mongooses: [mongoose], + logger +}); + +// store boolean if the job is cancelled +let isCancelled = false; + +// handle cancellation +if (parentPort) + parentPort.once('message', (message) => { + if (message === 'cancel') { + isCancelled = true; + } + }); + +graceful.listen(); + +/** + * Monitor queue for abuse patterns and apply throttling + */ +async function monitorQueueAbuse() { + if (isCancelled) return; + + try { + logger.info('Starting queue abuse monitoring...'); + + // Get users with high email volume in the past hour + const oneHourAgo = dayjs().subtract(1, 'hour').toDate(); + const suspiciousUsers = await Emails.aggregate([ + { + $match: { + created_at: { $gte: oneHourAgo }, + status: { $in: ['queued', 'sent', 'bounced', 'rejected'] } + } + }, + { + $group: { + _id: '$user', + totalEmails: { $sum: 1 }, + queuedEmails: { $sum: { $cond: [{ $eq: ['$status', 'queued'] }, 1, 0] } }, + bouncedEmails: { $sum: { $cond: [{ $eq: ['$status', 'bounced'] }, 1, 0] } }, + rejectedEmails: { $sum: { $cond: [{ $eq: ['$status', 'rejected'] }, 1, 0] } }, + avgPriority: { $avg: '$priority' } + } + }, + { + $match: { + $or: [ + { totalEmails: { $gte: 100 } }, // High volume + { queuedEmails: { $gte: 50 } }, // Many queued + { $expr: { $gte: [{ $divide: ['$bouncedEmails', '$totalEmails'] }, 0.1] } }, // 10%+ bounce rate + { $expr: { $gte: [{ $divide: ['$rejectedEmails', '$totalEmails'] }, 0.05] } } // 5%+ reject rate + ] + } + }, + { $sort: { totalEmails: -1 } }, + { $limit: 100 } // Check top 100 most active users + ]); + + logger.info(`Found ${suspiciousUsers.length} users requiring abuse score analysis`); + + let throttledUsers = 0; + let alertsSent = 0; + + // Analyze each suspicious user + for (const userStats of suspiciousUsers) { + if (isCancelled) break; + + try { + const userId = userStats._id; + const abuseScore = await fairQueue.calculateUserAbuseScore(userId); + + // Update the user's emails with current abuse score + await Emails.updateMany( + { user: userId, status: 'queued' }, + { $set: { abuse_score: abuseScore } } + ); + + logger.debug( + `User ${userId.toString().slice(-6)} - Volume: ${userStats.totalEmails}, Abuse Score: ${abuseScore}` + ); + + // Apply throttling if abuse score is too high + if (abuseScore >= PRIORITY_CONFIG.ABUSE_THRESHOLD) { + const throttleResult = await fairQueue.applyUserThrottling(userId, abuseScore); + + if (throttleResult.affectedEmails > 0) { + throttledUsers++; + + // Get user details for notification + const user = await Users.findById(userId) + .select(`id email ${config.lastLocaleField}`) + .lean(); + + if (user) { + // Send alert to admins + await emailHelper({ + template: 'alert', + message: { + to: config.email.message.from, + subject: 'User automatically throttled for suspicious activity' + }, + locals: { + message: ` +

User Throttled for Abuse

+

User: ${user.email} (${userId})

+

Abuse Score: ${abuseScore}/100

+

Throttled Until: ${throttleResult.throttledUntil.toISOString()}

+

Affected Emails: ${throttleResult.affectedEmails}

+

Duration: ${ms(throttleResult.duration, { long: true })}

+ +

Activity in Past Hour:

+ + ` + } + }); + + alertsSent++; + } + } + } + } catch (err) { + logger.error(`Error analyzing user ${userStats._id}:`, err); + } + } + + // Monitor overall queue fairness + const queueFairnessStats = await analyzeQueueFairness(); + + if (!queueFairnessStats.isFair) { + logger.warn('Queue fairness issue detected:', queueFairnessStats); + + // Send fairness alert if severe + if (queueFairnessStats.maxDomainPercentage > 50) { + await emailHelper({ + template: 'alert', + message: { + to: config.email.message.from, + subject: 'Queue fairness issue detected' + }, + locals: { + message: ` +

Queue Fairness Alert

+

A single domain is monopolizing the email queue.

+ + ` + } + }); + } + } + + logger.info( + `Abuse monitoring completed: ${throttledUsers} users throttled, ${alertsSent} alerts sent` + ); + + } catch (err) { + logger.error('Queue abuse monitoring error:', err); + } +} + +/** + * Analyze queue fairness metrics + */ +async function analyzeQueueFairness() { + const now = new Date(); + + // Get queue distribution by domain and user + const [domainStats, userStats, totalQueued] = await Promise.all([ + Emails.aggregate([ + { + $match: { + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: now } } + ] + } + }, + { + $group: { + _id: '$domain', + count: { $sum: 1 } + } + }, + { $sort: { count: -1 } } + ]), + + Emails.aggregate([ + { + $match: { + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: now } } + ] + } + }, + { + $group: { + _id: '$user', + count: { $sum: 1 } + } + }, + { $sort: { count: -1 } } + ]), + + Emails.countDocuments({ + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: now } } + ] + }) + ]); + + const maxDomainCount = domainStats[0]?.count || 0; + const maxUserCount = userStats[0]?.count || 0; + + const maxDomainPercentage = totalQueued > 0 ? (maxDomainCount / totalQueued) * 100 : 0; + const maxUserPercentage = totalQueued > 0 ? (maxUserCount / totalQueued) * 100 : 0; + + // Consider queue unfair if single domain/user has >30% of queue + const isFair = maxDomainPercentage <= 30 && maxUserPercentage <= 30; + + return { + totalQueued, + uniqueDomains: domainStats.length, + uniqueUsers: userStats.length, + maxDomainCount, + maxUserCount, + maxDomainPercentage, + maxUserPercentage, + isFair + }; +} + +/** + * Generate queue health report + */ +async function generateQueueHealthReport() { + try { + const queueHealth = await fairQueue.calculateQueueHealth(); + const fairnessStats = await analyzeQueueFairness(); + + // Get priority distribution + const priorityStats = await Emails.aggregate([ + { $match: { status: 'queued' } }, + { + $group: { + _id: '$priority', + count: { $sum: 1 }, + avgAbuseScore: { $avg: '$abuse_score' } + } + }, + { $sort: { _id: -1 } } + ]); + + // Get throttled users count + const throttledCount = await Emails.countDocuments({ + status: 'queued', + throttled_until: { $exists: true, $gt: new Date() } + }); + + const healthReport = { + timestamp: new Date(), + queueHealth, + fairnessStats, + priorityStats, + throttledCount + }; + + logger.info('Queue Health Report:', healthReport); + + return healthReport; + } catch (err) { + logger.error('Error generating queue health report:', err); + return null; + } +} + +(async () => { + await setupMongoose(logger); + + // Main monitoring loop + (async function startMonitoring() { + if (isCancelled) { + if (parentPort) parentPort.postMessage('done'); + else process.exit(0); + return; + } + + try { + // Run abuse monitoring every cycle + await monitorQueueAbuse(); + + // Generate health report (for debugging/monitoring) + await generateQueueHealthReport(); + + } catch (err) { + logger.error('Queue monitoring error:', err); + } + + // Wait 5 minutes before next check + setTimeout(startMonitoring, ms('5m')); + })(); +})(); \ No newline at end of file diff --git a/jobs/send-emails.js b/jobs/send-emails.js index 3ad687c99..132cd28a4 100644 --- a/jobs/send-emails.js +++ b/jobs/send-emails.js @@ -33,6 +33,8 @@ const processEmail = require('#helpers/process-email'); const setupMongoose = require('#helpers/setup-mongoose'); const getBlockedHashes = require('#helpers/get-blocked-hashes'); const monitorServer = require('#helpers/monitor-server'); +const fairQueue = require('#helpers/fair-queue'); +const { PRIORITY_LEVELS } = require('#config/priority-levels'); monitorServer(); @@ -83,156 +85,143 @@ async function sendEmails() { return; } - // TODO: capacity/recipient issues should be hard 550 bounce for outbound - - const now = new Date(); - const limit = config.smtpMaxQueue - queue.size; - - logger.info('queueing %d emails', limit); - - // TODO: filter out recently blocked targets by rejectedErrors[x].mx.target - - // - // NOTE: if you change this then also update `jobs/check-smtp-frozen-queue` if necessary - // - // get list of all suspended domains - // and recently blocked emails to exclude - const [suspendedDomains, recentlyBlocked] = await Promise.all([ - Domains.aggregate([ - { $match: { is_smtp_suspended: true } }, - { $group: { _id: '$_id' } } - ]) - .allowDiskUse(true) - .exec(), - Emails.aggregate([ - { - $match: { - updated_at: { - $gte: dayjs().subtract(1, 'hour').toDate(), - $lte: now - }, - has_blocked_hashes: true, - blocked_hashes: { - $in: getBlockedHashes(IP_ADDRESS) + try { + // Clean up expired throttling first + await fairQueue.cleanupExpiredThrottling(); + + // Calculate queue health and adaptive limits + const queueHealth = await fairQueue.calculateQueueHealth(); + const limit = config.smtpMaxQueue - queue.size; + + logger.info( + 'Fair queue: processing up to %d emails (health: %.2f, age: %dm)', + limit, + queueHealth.multiplier, + queueHealth.queueAge + ); + + if (limit <= 0) { + await setTimeout(5000); + return; + } + + const now = new Date(); + + // Get excluded domains and emails (same as original logic) + const [suspendedDomains, recentlyBlocked] = await Promise.all([ + Domains.aggregate([ + { $match: { is_smtp_suspended: true } }, + { $group: { _id: '$_id' } } + ]) + .allowDiskUse(true) + .exec(), + Emails.aggregate([ + { + $match: { + updated_at: { + $gte: dayjs().subtract(1, 'hour').toDate(), + $lte: now + }, + has_blocked_hashes: true, + blocked_hashes: { + $in: getBlockedHashes(IP_ADDRESS) + } } + }, + { + $group: { _id: '$_id' } } - }, - { - $group: { _id: '$_id' } - } - ]) - .allowDiskUse(true) - .exec() - ]); - - const suspendedDomainIds = suspendedDomains.map((v) => v._id); - const recentlyBlockedIds = recentlyBlocked.map((v) => v._id); - - logger.info('%d suspended domain ids', suspendedDomainIds.length); - - logger.info('%d recently blocked ids', recentlyBlockedIds.length); - - // - // TODO: warm up IP addresses - // - // - - // - // TODO: SMTP pooling by target - // - - // NOTE: if you change this then also update `jobs/check-smtp-frozen-queue` if necessary - // TODO: optimize this query - const query = { - _id: { $nin: recentlyBlockedIds }, - is_locked: false, - status: 'queued', - domain: { - $nin: suspendedDomainIds - }, - date: { - $lte: now + ]) + .allowDiskUse(true) + .exec() + ]); + + const suspendedDomainIds = suspendedDomains.map((v) => v._id); + const recentlyBlockedIds = recentlyBlocked.map((v) => v._id); + + logger.info('%d suspended domains, %d recently blocked emails', + suspendedDomainIds.length, recentlyBlockedIds.length); + + // Base query for fair queue processing + const query = { + _id: { $nin: recentlyBlockedIds }, + is_locked: false, + status: 'queued', + domain: { $nin: suspendedDomainIds }, + date: { $lte: now }, + // Exclude throttled emails (not expired) + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: now } } + ] + }; + + // Use fair distribution to get emails with adaptive priority limits + const selectedEmails = await fairQueue.getDomainFairDistribution(query, limit, queueHealth); + + logger.info(`Fair queue: selected ${selectedEmails.length} emails for processing`); + + // Process selected emails with priority-aware queuing + for (const email of selectedEmails) { + // return early if the job was already cancelled + if (isCancelled) break; + + // Add to queue with priority (PQueue doesn't use priority directly, + // but we maintain order through our selection process) + queue.add( + async () => { + try { + await processEmail({ email, resolver, client }); + } catch (err) { + logger.error(err, { email: email._id }); + } + }, + { + // Future: could implement priority queues here + priority: email.priority || PRIORITY_LEVELS.NORMAL + } + ); } - }; - - /* - const count = await Emails.countDocuments(query); - - logger.info('%d emails pending in queue', count); - - // if count is > limit then we need to aggregate - // and group for equal queue flow - if (count > limit) { - // TODO: get all those with priority > 0 first - // then subtract from limit and find more - // TODO: monetize priority queue feature (paid add-on) - - const domainIds = await Emails.distinct('domain', query); - // limit could be 60, domains 40 = 2 (1.5 -> 2) - // limit could be 30, domains = 50 = (0.6 -> 1) - // limit could be 2, domains = 1 = (2) - // limit could be 1, domains = 1000 = (0.001 -> 1) - const maxPerDomain = Math.ceil(limit / domainIds.length); - - const emailIds = []; - - // for each domain, subtract the count that it already has with "queued" + locked state - for (const domainId of domainIds) { - // eslint-disable-next-line no-await-in-loop - const ids = await Emails.distinct('_id', { - ...query, - domain: { $in: [domainId] }, - is_locked: true - }); - if (ids.length >= maxPerDomain) { - // cannot queue any more - logger.error( - 'Queue size exceeded for domain %s (%d/%d)', - domainId.toString(), - ids.length, - maxPerDomain - ); - } else { - logger.info( - 'Adding %d to queue for domain %s (%d/%d)', - maxPerDomain - ids.length, - domainId.toString(), - ids.length, - maxPerDomain - ); - emailIds.push(...ids.slice(0, maxPerDomain - ids.length)); + + // Log queue distribution for monitoring + if (selectedEmails.length > 0) { + const priorityDistribution = {}; + for (const email of selectedEmails) { + const priority = email.priority || PRIORITY_LEVELS.NORMAL; + priorityDistribution[priority] = (priorityDistribution[priority] || 0) + 1; } + + logger.debug('Priority distribution:', priorityDistribution); } - // modify `queue` such that it uses `_id: { $in: ids }` - // whereas `_id` is an email ID to send out - queue._id = { $in: emailIds }; - } - */ - - // eslint-disable-next-line unicorn/no-array-callback-reference - for await (const email of Emails.find(query) - .sort({ created_at: -1 }) // TODO: slows this query down by having sort - .lean() - .limit(limit) - .cursor() - .addCursorFlag('noCursorTimeout', true)) { - // return early if the job was already cancelled - if (isCancelled) break; - // TODO: implement queue on a per-target/provider basis (e.g. 10 at once to Cox addresses) - queue.add( - async () => { + } catch (err) { + logger.error('Fair queue error:', err); + // Fall back to basic processing in case of fair queue failure + const basicQuery = { + is_locked: false, + status: 'queued', + date: { $lte: new Date() } + }; + + const fallbackEmails = await Emails.find(basicQuery) + .sort({ priority: -1, created_at: 1 }) + .limit(Math.min(10, config.smtpMaxQueue - queue.size)) + .lean(); + + function processEmailWrapper(email) { + return async () => { try { await processEmail({ email, resolver, client }); } catch (err) { - logger.error(err, { email }); + logger.error(err, { email: email._id }); } - }, - { - // TODO: if the email was admin owned domain then priority higher (see email pre-save hook) - // priority: email.priority || 0 - } - ); + }; + } + + for (const email of fallbackEmails) { + if (isCancelled) break; + queue.add(processEmailWrapper(email)); + } } await setTimeout(5000); diff --git a/scripts/fair-queue-migration.js b/scripts/fair-queue-migration.js new file mode 100755 index 000000000..e21f13233 --- /dev/null +++ b/scripts/fair-queue-migration.js @@ -0,0 +1,210 @@ +#!/usr/bin/env node + +/** + * Copyright (c) Forward Email LLC + * SPDX-License-Identifier: BUSL-1.1 + */ + +// eslint-disable-next-line import/no-unassigned-import +require('#config/env'); + +const process = require('node:process'); + +// eslint-disable-next-line import/no-unassigned-import +require('#config/mongoose'); + +const Graceful = require('@ladjs/graceful'); +const mongoose = require('mongoose'); +const dayjs = require('dayjs-with-plugins'); + +const Emails = require('#models/emails'); +const Domains = require('#models/domains'); +const Users = require('#models/users'); +const logger = require('#helpers/logger'); +const setupMongoose = require('#helpers/setup-mongoose'); + +const graceful = new Graceful({ + mongooses: [mongoose], + logger +}); + +graceful.listen(); + +const BATCH_SIZE = 1000; + +// Priority levels constants +const PRIORITY_LEVELS = { + HIGH: 2, // Paid users, good reputation + NORMAL: 1, // Regular users (new default) + LOW: 0, // Free tier, new accounts + THROTTLED: -1 // Detected abuse/suspicious activity +}; + +async function migrateFairQueueFields() { + try { + logger.info('Starting fair queue migration...'); + + // Get total count for progress tracking + const totalCount = await Emails.countDocuments({}); + logger.info(`Total emails to migrate: ${totalCount}`); + + let migratedCount = 0; + let batchCount = 0; + + // Process emails in batches using cursor + const cursor = Emails.find({}).lean().cursor().addCursorFlag('noCursorTimeout', true); + + const batch = []; + + for (let email = await cursor.next(); email; email = await cursor.next()) { + batch.push(email); + + // Process batch when it reaches BATCH_SIZE or at the end + if (batch.length >= BATCH_SIZE || (!await cursor.next() && batch.length > 0)) { + batchCount++; + logger.info(`Processing batch ${batchCount} (${batch.length} emails)...`); + + const bulkOps = []; + + for (const email of batch) { + const updateFields = {}; + let needsUpdate = false; + + // Set default priority to NORMAL (1) if currently 0 or missing + if (email.priority === 0 || email.priority === undefined) { + updateFields.priority = PRIORITY_LEVELS.NORMAL; + needsUpdate = true; + } + + // Set default abuse_score if missing + if (email.abuse_score === undefined) { + updateFields.abuse_score = 0; + needsUpdate = true; + } + + // Determine priority based on domain/user info + if (email.domain && email.user) { + try { + const [domain, user] = await Promise.all([ + Domains.findById(email.domain).lean().populate('members.user', 'id group plan created_at'), + Users.findById(email.user).lean().select('id plan created_at group') + ]); + + if (domain && user) { + let newPriority = PRIORITY_LEVELS.NORMAL; + + // Check if user is an admin or domain has admin users with premium plans + const isAdmin = user.group === 'admin'; + const adminExists = domain.members.some( + m => m.user && m.user.group === 'admin' && + ['enhanced_protection', 'team'].includes(m.user.plan) + ); + + if (isAdmin || adminExists) { + newPriority = PRIORITY_LEVELS.HIGH; + } else if (dayjs().diff(user.created_at, 'days') < 7) { + // New accounts get lower priority + newPriority = PRIORITY_LEVELS.LOW; + } + + if (email.priority !== newPriority) { + updateFields.priority = newPriority; + needsUpdate = true; + } + } + } catch (err) { + logger.warn(`Failed to lookup domain/user for email ${email._id}: ${err.message}`); + // Continue with default priority + } + } + + if (needsUpdate) { + bulkOps.push({ + updateOne: { + filter: { _id: email._id }, + update: { $set: updateFields } + } + }); + } + } + + // Execute batch updates + if (bulkOps.length > 0) { + const result = await Emails.bulkWrite(bulkOps, { ordered: false }); + logger.info(`Updated ${result.modifiedCount} emails in batch ${batchCount}`); + migratedCount += result.modifiedCount; + } + + batch.length = 0; // Clear batch + + // Progress update + const processed = Math.min(batchCount * BATCH_SIZE, totalCount); + const percentage = ((processed / totalCount) * 100).toFixed(1); + logger.info(`Progress: ${processed}/${totalCount} (${percentage}%)`); + } + } + + logger.info(`Migration completed! Updated ${migratedCount} emails total.`); + + // Create indexes after migration (in case they don't exist) + logger.info('Ensuring indexes exist...'); + + await Promise.all([ + // Fair queue composite index + Emails.collection.createIndex({ + status: 1, + priority: -1, + domain: 1, + user: 1, + created_at: 1 + }, { background: true }), + + // Throttled emails index + Emails.collection.createIndex( + { throttled_until: 1, status: 1 }, + { + background: true, + partialFilterExpression: { throttled_until: { $exists: true } } + } + ), + + // Abuse score index + Emails.collection.createIndex( + { abuse_score: 1, user: 1 }, + { background: true } + ) + ]); + + logger.info('Indexes created successfully!'); + + // Generate migration summary + const priorityCounts = await Emails.aggregate([ + { $group: { _id: '$priority', count: { $sum: 1 } } }, + { $sort: { _id: 1 } } + ]); + + logger.info('Priority distribution after migration:'); + for (const { _id, count } of priorityCounts) { + const levelName = Object.keys(PRIORITY_LEVELS).find(k => PRIORITY_LEVELS[k] === _id) || 'UNKNOWN'; + logger.info(` Priority ${_id} (${levelName}): ${count} emails`); + } + + logger.info('Fair queue migration completed successfully!'); + + } catch (err) { + logger.error('Migration failed:', err); + throw err; + } +} + +(async () => { + await setupMongoose(logger); + + try { + await migrateFairQueueFields(); + process.exit(0); + } catch (err) { + logger.fatal(err); + process.exit(1); + } +})(); \ No newline at end of file diff --git a/scripts/test-fair-queue-logic.js b/scripts/test-fair-queue-logic.js new file mode 100755 index 000000000..3f9e1f726 --- /dev/null +++ b/scripts/test-fair-queue-logic.js @@ -0,0 +1,489 @@ +#!/usr/bin/env node + +/** + * Copyright (c) Forward Email LLC + * SPDX-License-Identifier: BUSL-1.1 + */ + +// eslint-disable-next-line import/no-unassigned-import +require('#config/env'); + +const process = require('node:process'); + +// eslint-disable-next-line import/no-unassigned-import +require('#config/mongoose'); + +const Graceful = require('@ladjs/graceful'); +const mongoose = require('mongoose'); +const dayjs = require('dayjs-with-plugins'); + +const Emails = require('#models/emails'); +const Users = require('#models/users'); +const Domains = require('#models/domains'); +const logger = require('#helpers/logger'); +const setupMongoose = require('#helpers/setup-mongoose'); +const fairQueue = require('#helpers/fair-queue'); +const { PRIORITY_LEVELS } = require('#config/priority-levels'); + +const graceful = new Graceful({ + mongooses: [mongoose], + logger +}); + +graceful.listen(); + +/** + * Create test data for fair queue testing + */ +async function createTestData() { + logger.info('Creating test data...'); + + // Clean up any existing test data + await Promise.all([ + Emails.deleteMany({ messageId: /^test-fair-queue/ }), + Users.deleteMany({ email: /^test-fair-queue/ }), + Domains.deleteMany({ name: /^test-fair-queue/ }) + ]); + + // Create test users + const users = []; + const domains = []; + + // Create 3 domains with different characteristics + for (let i = 1; i <= 3; i++) { + const domain = new Domains({ + name: `test-fair-queue-domain${i}.com`, + plan: i === 1 ? 'team' : 'free', // Domain 1 is premium + members: [] + }); + await domain.save(); + domains.push(domain); + + // Create 2-3 users per domain + const usersPerDomain = i === 2 ? 3 : 2; // Domain 2 has more users + for (let j = 1; j <= usersPerDomain; j++) { + const isAdmin = i === 1 && j === 1; // First user of first domain is admin + const isPremium = i === 1 || (i === 3 && j === 1); // Premium users + const isNew = i === 3 && j === 2; // New user (created recently) + + const user = new Users({ + email: `test-fair-queue-user${i}-${j}@example.com`, + password: 'testpassword123', + plan: isPremium ? 'team' : 'free', + group: isAdmin ? 'admin' : 'user', + created_at: isNew ? new Date() : dayjs().subtract(30, 'days').toDate(), + [require('#config').userFields.hasVerifiedEmail]: true + }); + await user.save(); + users.push(user); + + // Add user to domain + domain.members.push({ + user: user._id, + group: isAdmin ? 'admin' : 'user' + }); + } + await domain.save(); + } + + // Create test emails with various scenarios + const emails = []; + let emailCounter = 0; + + for (const domain of domains) { + const domainUsers = users.filter(u => + domain.members.some(m => m.user.equals(u._id)) + ); + + for (const user of domainUsers) { + // Create different numbers of emails per user to test fairness + let emailCount; + if (domain.name.includes('domain1')) { + emailCount = user.group === 'admin' ? 5 : 3; // Admin gets more emails + } else if (domain.name.includes('domain2')) { + emailCount = 10; // High volume domain + } else { + emailCount = user.created_at > dayjs().subtract(7, 'days').toDate() ? 2 : 4; // New users get fewer + } + + for (let i = 0; i < emailCount; i++) { + emailCounter++; + const email = new Emails({ + messageId: `test-fair-queue-${emailCounter}`, + status: 'queued', + user: user._id, + domain: domain._id, + envelope: { + from: user.email, + to: [`recipient${emailCounter}@example.com`] + }, + message: `Test email ${emailCounter} content`, + headers: { + subject: `Test Subject ${emailCounter}`, + from: user.email, + to: `recipient${emailCounter}@example.com`, + date: new Date().toISOString() + }, + date: dayjs().subtract(Math.random() * 60, 'minutes').toDate(), // Random times in past hour + created_at: dayjs().subtract(Math.random() * 60, 'minutes').toDate() + }); + + // Priorities will be set automatically by pre-save hook + await email.save(); + emails.push(email); + } + } + } + + logger.info(`Created test data: ${domains.length} domains, ${users.length} users, ${emails.length} emails`); + + return { domains, users, emails }; +} + +/** + * Test queue health calculation + */ +async function testQueueHealth() { + logger.info('Testing queue health calculation...'); + + const queueHealth = await fairQueue.calculateQueueHealth(); + + logger.info('Queue health metrics:', queueHealth); + + // Validate metrics + if (typeof queueHealth.totalQueued !== 'number') { + throw new Error('totalQueued should be a number'); + } + if (typeof queueHealth.queueAge !== 'number') { + throw new Error('queueAge should be a number'); + } + if (typeof queueHealth.multiplier !== 'number' || queueHealth.multiplier < 0.1 || queueHealth.multiplier > 2.0) { + throw new Error('multiplier should be between 0.1 and 2.0'); + } + if (typeof queueHealth.isHealthy !== 'boolean') { + throw new Error('isHealthy should be a boolean'); + } + + logger.info('✓ Queue health calculation works correctly'); + return queueHealth; +} + +/** + * Test adaptive limits calculation + */ +async function testAdaptiveLimits(queueHealth) { + logger.info('Testing adaptive limits calculation...'); + + const baseLimit = 50; + const limits = fairQueue.calculateAdaptiveLimits(baseLimit, queueHealth); + + logger.info('Adaptive limits:', limits); + + // Validate limits + for (const [priority, limit] of Object.entries(limits)) { + if (typeof limit !== 'number' || limit < 0) { + throw new Error(`Limit for priority ${priority} should be a non-negative number`); + } + } + + // Check that we have limits for all priority levels + const expectedPriorities = Object.values(PRIORITY_LEVELS); + for (const priority of expectedPriorities) { + if (!(priority in limits)) { + throw new Error(`Missing limit for priority ${priority}`); + } + } + + // Total allocated should not exceed base limit significantly + const totalAllocated = Object.values(limits).reduce((sum, limit) => sum + limit, 0); + if (totalAllocated > baseLimit * 1.2) { // Allow 20% over for rounding + throw new Error(`Total allocated (${totalAllocated}) exceeds base limit (${baseLimit}) by too much`); + } + + logger.info('✓ Adaptive limits calculation works correctly'); + return limits; +} + +/** + * Test domain fair distribution + */ +async function testDomainFairDistribution(queueHealth) { + logger.info('Testing domain fair distribution...'); + + const query = { + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: new Date() } } + ] + }; + + const limit = 20; + + // Test without queue health (basic fair distribution) + const basicEmails = await fairQueue.getDomainFairDistribution(query, limit); + logger.info(`Basic distribution selected ${basicEmails.length} emails`); + + // Test with queue health (adaptive distribution) + const adaptiveEmails = await fairQueue.getDomainFairDistribution(query, limit, queueHealth); + logger.info(`Adaptive distribution selected ${adaptiveEmails.length} emails`); + + // Analyze distribution fairness + const basicDomainCounts = {}; + const adaptiveDomainCounts = {}; + const basicPriorityCounts = {}; + const adaptivePriorityCounts = {}; + + for (const email of basicEmails) { + const domainKey = email.domain.toString().slice(-6); + basicDomainCounts[domainKey] = (basicDomainCounts[domainKey] || 0) + 1; + basicPriorityCounts[email.priority] = (basicPriorityCounts[email.priority] || 0) + 1; + } + + for (const email of adaptiveEmails) { + const domainKey = email.domain.toString().slice(-6); + adaptiveDomainCounts[domainKey] = (adaptiveDomainCounts[domainKey] || 0) + 1; + adaptivePriorityCounts[email.priority] = (adaptivePriorityCounts[email.priority] || 0) + 1; + } + + logger.info('Basic distribution by domain:', basicDomainCounts); + logger.info('Adaptive distribution by domain:', adaptiveDomainCounts); + logger.info('Basic distribution by priority:', basicPriorityCounts); + logger.info('Adaptive distribution by priority:', adaptivePriorityCounts); + + // Validate fairness (no single domain should have >60% of emails) + const basicMaxDomain = Math.max(...Object.values(basicDomainCounts)); + const adaptiveMaxDomain = Math.max(...Object.values(adaptiveDomainCounts)); + + if (basicMaxDomain / basicEmails.length > 0.6) { + logger.warn('Basic distribution may not be fair - single domain has >60% of emails'); + } + if (adaptiveMaxDomain / adaptiveEmails.length > 0.6) { + logger.warn('Adaptive distribution may not be fair - single domain has >60% of emails'); + } + + logger.info('✓ Domain fair distribution works correctly'); + + return { basicEmails, adaptiveEmails }; +} + +/** + * Test abuse score calculation + */ +async function testAbuseScoreCalculation() { + logger.info('Testing abuse score calculation...'); + + // Get a test user + const testUser = await Users.findOne({ email: /test-fair-queue/ }).lean(); + if (!testUser) { + throw new Error('No test user found'); + } + + const abuseScore = await fairQueue.calculateUserAbuseScore(testUser._id); + + logger.info(`User abuse score: ${abuseScore}`); + + // Validate score + if (typeof abuseScore !== 'number' || abuseScore < 0 || abuseScore > 100) { + throw new Error('Abuse score should be between 0 and 100'); + } + + logger.info('✓ Abuse score calculation works correctly'); + return abuseScore; +} + +/** + * Test throttling functionality + */ +async function testThrottling() { + logger.info('Testing throttling functionality...'); + + // Get a test user with multiple emails + const testUser = await Users.findOne({ email: /test-fair-queue/ }).lean(); + if (!testUser) { + throw new Error('No test user found'); + } + + const beforeCount = await Emails.countDocuments({ + user: testUser._id, + status: 'queued', + throttled_until: { $exists: false } + }); + + logger.info(`User has ${beforeCount} unthrottled emails before throttling`); + + // Apply throttling with a high abuse score + const throttleResult = await fairQueue.applyUserThrottling(testUser._id, 75); + + logger.info('Throttle result:', throttleResult); + + const afterCount = await Emails.countDocuments({ + user: testUser._id, + status: 'queued', + throttled_until: { $exists: false } + }); + + logger.info(`User has ${afterCount} unthrottled emails after throttling`); + + // Validate throttling was applied + if (throttleResult.affectedEmails === 0) { + logger.warn('No emails were throttled - this may be expected if user had no queued emails'); + } else if (afterCount >= beforeCount) { + throw new Error('Throttling did not reduce unthrottled email count'); + } + + // Test cleanup of expired throttling + logger.info('Testing throttling cleanup...'); + + // Manually expire throttling for testing + await Emails.updateMany( + { user: testUser._id, throttled_until: { $exists: true } }, + { $set: { throttled_until: new Date(Date.now() - 1000) } } // 1 second ago + ); + + const cleanedUp = await fairQueue.cleanupExpiredThrottling(); + logger.info(`Cleaned up throttling for ${cleanedUp} emails`); + + logger.info('✓ Throttling functionality works correctly'); + return throttleResult; +} + +/** + * Test end-to-end fair queue scenario + */ +async function testEndToEndScenario() { + logger.info('Testing end-to-end fair queue scenario...'); + + // Simulate the main send-emails job logic + const queueHealth = await fairQueue.calculateQueueHealth(); + const limit = 15; + + const query = { + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: new Date() } } + ] + }; + + logger.info(`Simulating queue processing with limit=${limit}, health=${queueHealth.multiplier.toFixed(2)}`); + + // Clean up expired throttling + await fairQueue.cleanupExpiredThrottling(); + + // Get fair distribution + const selectedEmails = await fairQueue.getDomainFairDistribution(query, limit, queueHealth); + + logger.info(`Selected ${selectedEmails.length} emails for processing`); + + // Analyze the selection + const domainCounts = {}; + const userCounts = {}; + const priorityCounts = {}; + + for (const email of selectedEmails) { + const domainKey = email.domain.toString().slice(-6); + const userKey = email.user.toString().slice(-6); + + domainCounts[domainKey] = (domainCounts[domainKey] || 0) + 1; + userCounts[userKey] = (userCounts[userKey] || 0) + 1; + priorityCounts[email.priority] = (priorityCounts[email.priority] || 0) + 1; + } + + logger.info('End-to-end distribution by domain:', domainCounts); + logger.info('End-to-end distribution by user:', userCounts); + logger.info('End-to-end distribution by priority:', priorityCounts); + + // Calculate fairness metrics + const totalEmails = selectedEmails.length; + const maxDomainPercentage = Math.max(...Object.values(domainCounts)) / totalEmails * 100; + const maxUserPercentage = Math.max(...Object.values(userCounts)) / totalEmails * 100; + + logger.info(`Fairness metrics: Max domain: ${maxDomainPercentage.toFixed(1)}%, Max user: ${maxUserPercentage.toFixed(1)}%`); + + // Validate fairness + if (maxDomainPercentage > 70) { + logger.warn(`Domain distribution may be unfair: ${maxDomainPercentage.toFixed(1)}% for single domain`); + } + if (maxUserPercentage > 50) { + logger.warn(`User distribution may be unfair: ${maxUserPercentage.toFixed(1)}% for single user`); + } + + logger.info('✓ End-to-end scenario completed successfully'); + + return { + selectedEmails, + domainCounts, + userCounts, + priorityCounts, + fairnessMetrics: { + maxDomainPercentage, + maxUserPercentage + } + }; +} + +/** + * Clean up test data + */ +async function cleanupTestData() { + logger.info('Cleaning up test data...'); + + const deleteResults = await Promise.all([ + Emails.deleteMany({ messageId: /^test-fair-queue/ }), + Users.deleteMany({ email: /^test-fair-queue/ }), + Domains.deleteMany({ name: /^test-fair-queue/ }) + ]); + + logger.info(`Cleaned up: ${deleteResults[0].deletedCount} emails, ${deleteResults[1].deletedCount} users, ${deleteResults[2].deletedCount} domains`); +} + +/** + * Main test function + */ +async function runFairQueueTests() { + try { + logger.info('Starting fair queue logic tests...'); + + // Create test data + await createTestData(); + + // Run individual tests + const queueHealth = await testQueueHealth(); + await testAdaptiveLimits(queueHealth); + await testDomainFairDistribution(queueHealth); + await testAbuseScoreCalculation(); + await testThrottling(); + + // Run end-to-end test + const endToEndResult = await testEndToEndScenario(); + + logger.info('All fair queue tests passed successfully!'); + logger.info('Test Summary:', { + queueHealth: queueHealth.isHealthy, + emailsProcessed: endToEndResult.selectedEmails.length, + fairnessMetrics: endToEndResult.fairnessMetrics + }); + + return true; + + } catch (err) { + logger.error('Fair queue test failed:', err); + return false; + } finally { + // Always clean up test data + await cleanupTestData(); + } +} + +(async () => { + await setupMongoose(logger); + + try { + const success = await runFairQueueTests(); + process.exit(success ? 0 : 1); + } catch (err) { + logger.fatal(err); + process.exit(1); + } +})(); \ No newline at end of file diff --git a/scripts/test-fair-queue-schema.js b/scripts/test-fair-queue-schema.js new file mode 100755 index 000000000..3f8fc84a7 --- /dev/null +++ b/scripts/test-fair-queue-schema.js @@ -0,0 +1,234 @@ +#!/usr/bin/env node + +/** + * Copyright (c) Forward Email LLC + * SPDX-License-Identifier: BUSL-1.1 + */ + +// eslint-disable-next-line import/no-unassigned-import +require('#config/env'); + +const process = require('node:process'); + +// eslint-disable-next-line import/no-unassigned-import +require('#config/mongoose'); + +const Graceful = require('@ladjs/graceful'); +const mongoose = require('mongoose'); + +const Emails = require('#models/emails'); +const logger = require('#helpers/logger'); +const setupMongoose = require('#helpers/setup-mongoose'); +const { PRIORITY_LEVELS } = require('#config/priority-levels'); + +const graceful = new Graceful({ + mongooses: [mongoose], + logger +}); + +graceful.listen(); + +async function testSchemaChanges() { + try { + logger.info('Testing fair queue schema changes...'); + + // Test 1: Validate schema fields exist and have correct defaults + logger.info('Test 1: Schema validation...'); + + const schema = Emails.schema; + const paths = schema.paths; + + // Check priority field + if (!paths.priority) { + throw new Error('Priority field missing from schema'); + } + if (paths.priority.options.default !== PRIORITY_LEVELS.NORMAL) { + throw new Error(`Priority default should be ${PRIORITY_LEVELS.NORMAL}, got ${paths.priority.options.default}`); + } + logger.info('✓ Priority field configured correctly'); + + // Check abuse_score field + if (!paths.abuse_score) { + throw new Error('abuse_score field missing from schema'); + } + if (paths.abuse_score.options.default !== 0) { + throw new Error('abuse_score default should be 0'); + } + if (paths.abuse_score.options.min !== 0 || paths.abuse_score.options.max !== 100) { + throw new Error('abuse_score should have min: 0, max: 100'); + } + logger.info('✓ abuse_score field configured correctly'); + + // Check throttled_until field + if (!paths.throttled_until) { + throw new Error('throttled_until field missing from schema'); + } + if (paths.throttled_until.instance !== 'Date') { + throw new Error('throttled_until should be Date type'); + } + logger.info('✓ throttled_until field configured correctly'); + + // Test 2: Check indexes exist + logger.info('Test 2: Index validation...'); + + const indexes = await Emails.collection.getIndexes(); + logger.info(`Found ${Object.keys(indexes).length} indexes`); + + // Check for fair queue composite index + const fairQueueIndex = Object.keys(indexes).find(key => + indexes[key].some(field => + typeof field === 'object' && + field.status === 1 && + field.priority === -1 && + field.domain === 1 && + field.user === 1 && + field.created_at === 1 + ) + ); + + if (fairQueueIndex) { + logger.info('✓ Fair queue composite index exists'); + } else { + logger.warn('Fair queue composite index not found - may need to run migration'); + } + + // Test 3: Test creating email with new fields + logger.info('Test 3: Email creation test...'); + + // Note: We'll do a dry run without actually saving to avoid dependencies + const testEmail = new Emails({ + priority: PRIORITY_LEVELS.HIGH, + abuse_score: 25, + throttled_until: new Date(Date.now() + 60000), // 1 minute from now + status: 'queued', + envelope: { from: 'test@example.com', to: ['recipient@example.com'] }, + message: 'Test message content', + messageId: 'test-message-id', + headers: { subject: 'Test Subject', from: 'test@example.com' }, + date: new Date(), + user: new mongoose.Types.ObjectId(), + domain: new mongoose.Types.ObjectId() + }); + + // Validate without saving + await testEmail.validate(); + logger.info('✓ Email creation with new fields validates successfully'); + + // Test 4: Test priority validation + logger.info('Test 4: Priority validation test...'); + + const testPriorities = [ + PRIORITY_LEVELS.THROTTLED, + PRIORITY_LEVELS.LOW, + PRIORITY_LEVELS.NORMAL, + PRIORITY_LEVELS.HIGH, + 5 // Custom high priority + ]; + + for (const priority of testPriorities) { + const email = new Emails({ + priority, + status: 'queued', + envelope: { from: 'test@example.com', to: ['recipient@example.com'] }, + message: 'Test message', + messageId: `test-${priority}`, + headers: { subject: 'Test', from: 'test@example.com' }, + date: new Date(), + user: new mongoose.Types.ObjectId(), + domain: new mongoose.Types.ObjectId() + }); + + await email.validate(); + if (email.priority !== priority) { + throw new Error(`Priority validation failed for ${priority}`); + } + } + logger.info('✓ Priority validation works for all levels'); + + // Test 5: Test abuse score validation + logger.info('Test 5: Abuse score validation test...'); + + // Valid abuse scores + for (const score of [0, 25, 50, 75, 100]) { + const email = new Emails({ + abuse_score: score, + status: 'queued', + envelope: { from: 'test@example.com', to: ['recipient@example.com'] }, + message: 'Test message', + messageId: `test-abuse-${score}`, + headers: { subject: 'Test', from: 'test@example.com' }, + date: new Date(), + user: new mongoose.Types.ObjectId(), + domain: new mongoose.Types.ObjectId() + }); + + await email.validate(); + if (email.abuse_score !== score) { + throw new Error(`Abuse score validation failed for ${score}`); + } + } + logger.info('✓ Abuse score validation works for valid ranges'); + + // Test invalid abuse scores should fail + try { + const invalidEmail = new Emails({ + abuse_score: 150, // Invalid: > 100 + status: 'queued', + envelope: { from: 'test@example.com', to: ['recipient@example.com'] }, + message: 'Test message', + messageId: 'test-invalid-abuse', + headers: { subject: 'Test', from: 'test@example.com' }, + date: new Date(), + user: new mongoose.Types.ObjectId(), + domain: new mongoose.Types.ObjectId() + }); + + await invalidEmail.validate(); + throw new Error('Should have failed validation for abuse_score > 100'); + } catch (err) { + if (err.message.includes('validation failed')) { + logger.info('✓ Invalid abuse score properly rejected'); + } else { + throw err; + } + } + + // Test 6: Sample queries that would be used by fair queue + logger.info('Test 6: Sample fair queue queries...'); + + const sampleQuery = { + status: 'queued', + $or: [ + { throttled_until: { $exists: false } }, + { throttled_until: { $lt: new Date() } } + ] + }; + + // This query should work with our new indexes + await Emails.find(sampleQuery) + .sort({ priority: -1, created_at: 1 }) + .limit(10) + .explain(sampleQuery); + + logger.info('✓ Fair queue query execution plan generated successfully'); + + logger.info('All tests passed! Schema changes are working correctly.'); + + } catch (err) { + logger.error('Schema test failed:', err); + throw err; + } +} + +(async () => { + await setupMongoose(logger); + + try { + await testSchemaChanges(); + logger.info('Schema testing completed successfully!'); + process.exit(0); + } catch (err) { + logger.fatal(err); + process.exit(1); + } +})(); \ No newline at end of file