Kafka Job Orchestrator (KJO)

const { AbstractProducer } = require('@jonaskahn/kafka-job-orchestrator');

class EmailProducer extends AbstractProducer {
    static get STATES() {
        return {
            PENDING: 1,
            COMPLETED: 2,
            FAILED: 3
        };
    }

    constructor() {
        const config = {
            // REQUIRED: Kafka topic name
            topic: 'email-queue',
            
            // REQUIRED: Redis configuration
            redisOptions: {
                // REQUIRED: Redis key prefix for deduplication
                keyPrefix: 'EMAIL',
                
                // OPTIONAL: TTL for processing keys (default: 300 seconds)
                processingTtl: 300,
                
                // OPTIONAL: TTL for sent message keys (default: 20 seconds)
                sentTtl: 20
            },
            
            // OPTIONAL: Topic auto-creation configuration
            topicOptions: {
                // OPTIONAL: Number of partitions (default: 1)
                partitions: 4,
                
                // OPTIONAL: Replication factor (default: 1)
                replicationFactor: 2,
                
                // OPTIONAL: Auto-create topic if not exists (default: true)
                autoCreate: true,
                
                // OPTIONAL: Additional Kafka topic configurations
                configEntries: {
                    'retention.ms': '604800000', // 7 days retention
                    'compression.type': 'gzip',
                    'cleanup.policy': 'delete'
                },
                
                // OPTIONAL: Additional topic options
                retentionMs: '604800000',
                segmentMs: '86400000', // 1 day
                compressionType: 'gzip',
                cleanupPolicy: 'delete'
            }
        };
        super(config);
    }

    // REQUIRED: Fetch emails from your database
    async getNextProcessingItems(criteria, limit, excludedIds) {
        const { state, priority } = criteria;
        
        return await EmailModel.find({
            state: state,
            priority: priority || { $exists: true },
            _id: { $nin: excludedIds }
        })
        .sort({ priority: -1, createdAt: 1 })
        .limit(limit);
    }

    // REQUIRED: Extract email ID
    getItemId(email) {
        return email._id.toString();
    }

    // REQUIRED: Extract display key for logging
    getItemKey(email) {
        return email.recipient;
    }

    // OPTIONAL: Called on successful message sending
    async onSuccess(emailId) {
        console.log(`✅ Successfully sent email: ${emailId}`);
        // Add custom success handling, metrics, etc.
    }

    // OPTIONAL: Called on failed message sending
    async onFailed(emailId, error) {
        console.error(`❌ Failed to send email: ${emailId}`, error);
        // Add custom error handling, retry logic, etc.
    }

    // Convenience methods for different use cases
    async producePendingEmails(limit = 100) {
        return await this.produceMessages(
            { state: EmailProducer.STATES.PENDING, priority: 'high' },
            limit,
            'new'
        );
    }

    async produceFailedEmails(limit = 50) {
        return await this.produceMessages(
            { state: EmailProducer.STATES.FAILED },
            limit,
            'retry'
        );
    }

    async produceHighPriorityEmails(limit = 50) {
        return await this.produceMessages(
            { state: EmailProducer.STATES.PENDING, priority: { $gte: 8 } },
            limit,
            'urgent'
        );
    }
}

const cron = require('node-cron');
const EmailProducer = require('./EmailProducer');

class EmailScheduler {
    constructor() {
        this.producer = new EmailProducer();
        this.isRunning = false;
        this.isRetryRunning = false;
    }

    async start() {
        await this.producer.connect();
        
        // Schedule new emails every minute
        cron.schedule('* * * * *', async () => {
            if (this.isRunning) {
                console.log('⏭️ Skipping - producer already running');
                return;
            }
            
            this.isRunning = true;
            try {
                console.log('📧 Checking for pending emails...');
                
                // Process high priority emails first
                const urgentResult = await this.producer.produceHighPriorityEmails(25);
                console.log(`🚨 Produced ${urgentResult?.messages?.length || 0} urgent emails`);
                
                // Process regular pending emails
                const result = await this.producer.producePendingEmails(75);
                console.log(`✅ Produced ${result?.messages?.length || 0} regular emails`);
                
            } catch (error) {
                console.error('❌ Email production failed:', error);
                // Optional: Add alerting/monitoring here
            } finally {
                this.isRunning = false;
            }
        });

        // Retry failed emails every 5 minutes
        cron.schedule('*/5 * * * *', async () => {
            if (this.isRetryRunning) {
                console.log('⏭️ Skipping retry - already running');
                return;
            }
            
            this.isRetryRunning = true;
            try {
                console.log('🔄 Retrying failed emails...');
                const retryResult = await this.producer.produceFailedEmails(20);
                console.log(`♻️ Retried ${retryResult?.messages?.length || 0} failed emails`);
            } catch (error) {
                console.error('❌ Failed email retry error:', error);
                // Optional: Add alerting for retry failures
            } finally {
                this.isRetryRunning = false;
            }
        });

        // Health check - every 10 minutes
        cron.schedule('*/10 * * * *', async () => {
            try {
                // Check system health
                const stats = {
                    producer: this.producer.isConnected,
                    currentlyRunning: this.isRunning,
                    retryRunning: this.isRetryRunning,
                    timestamp: new Date().toISOString()
                };
                console.log('💓 Health check:', stats);
            } catch (error) {
                console.error('❌ Health check failed:', error);
            }
        });

        console.log('📅 Email scheduler started with cron jobs:');
        console.log('  • Pending emails: every minute');
        console.log('  • Failed retries: every 5 minutes');
        console.log('  • Health checks: every 10 minutes');
    }

    async stop() {
        console.log('🛑 Stopping email scheduler...');
        
        // Wait for current operations to complete
        while (this.isRunning || this.isRetryRunning) {
            console.log('⏳ Waiting for operations to complete...');
            await new Promise(resolve => setTimeout(resolve, 1000));
        }
        
        // Disconnect producer
        await this.producer.disconnect();
        console.log('📅 Email scheduler stopped');
    }

    // Manual trigger methods for testing/debugging
    async triggerPendingEmails(limit = 100) {
        console.log('🔧 Manual trigger: pending emails');
        return await this.producer.producePendingEmails(limit);
    }

    async triggerFailedEmails(limit = 50) {
        console.log('🔧 Manual trigger: failed emails');
        return await this.producer.produceFailedEmails(limit);
    }
}

// Usage
const scheduler = new EmailScheduler();

async main() {
    try {
        await scheduler.start();
        
        // Keep the process alive
        process.on('SIGINT', async () => {
            console.log('\\n🔄 Received SIGINT, shutting down gracefully...');
            await scheduler.stop();
            process.exit(0);
        });

        process.on('SIGTERM', async () => {
            console.log('\\n🔄 Received SIGTERM, shutting down gracefully...');
            await scheduler.stop();
            process.exit(0);
        });

        // Handle uncaught errors
        process.on('unhandledRejection', (reason, promise) => {
            console.error('💥 Unhandled Rejection at:', promise, 'reason:', reason);
        });

        process.on('uncaughtException', (error) => {
            console.error('💥 Uncaught Exception:', error);
            process.exit(1);
        });

    } catch (error) {
        console.error('💥 Failed to start scheduler:', error);
        process.exit(1);
    }
}

// Start the scheduler
main();