diff --git a/email_service/src/email/email.service.ts b/email_service/src/email/email.service.ts index e6eb490..32bcec8 100644 --- a/email_service/src/email/email.service.ts +++ b/email_service/src/email/email.service.ts @@ -6,6 +6,7 @@ import * as Handlebars from 'handlebars'; import * as amqp from 'amqp-connection-manager'; import { ChannelWrapper } from 'amqp-connection-manager'; import { createClient, RedisClientType } from 'redis'; +import { ConsumeMessage } from 'amqplib'; enum NotificationStatus { PENDING = 'pending', @@ -22,6 +23,7 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { private transporter: nodemailer.Transporter; private retryAttempts = new Map(); private redisClient: RedisClientType; + private isProcessing = false; constructor(private readonly httpService: HttpService) { this.transporter = nodemailer.createTransport({ @@ -36,11 +38,26 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { } async onModuleInit() { + console.log('šŸš€ Email Service initializing...'); + // Connect to Redis await this.connectRedis(); // Connect to RabbitMQ await this.connectRabbitMQ(); + + // Verify SMTP connection + await this.verifySmtpConnection(); + } + + private async verifySmtpConnection() { + try { + await this.transporter.verify(); + console.log('āœ… SMTP connection verified'); + } catch (error) { + console.error('āŒ SMTP connection failed:', error.message); + console.error('Check your SMTP_USER and SMTP_PASS environment variables'); + } } private async connectRedis() { @@ -51,15 +68,27 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { socket: { host: redisHost, port: redisPort, + reconnectStrategy: (retries) => { + if (retries > 10) { + console.error('āŒ Redis max reconnection attempts reached'); + return new Error('Max reconnection attempts reached'); + } + const delay = Math.min(retries * 100, 3000); + console.log(`šŸ”„ Reconnecting to Redis... (attempt ${retries})`); + return delay; + }, }, password: process.env.REDIS_PASSWORD, }); this.redisClient.on('error', (err) => - console.error('Redis Client Error', err), + console.error('āŒ Redis Client Error:', err.message), ); this.redisClient.on('connect', () => - console.log('Email Service: Redis connected'), + console.log('āœ… Email Service: Redis connected'), + ); + this.redisClient.on('reconnecting', () => + console.log('šŸ”„ Redis reconnecting...'), ); await this.redisClient.connect(); @@ -69,42 +98,137 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { const rabbitMQUrl = process.env.RABBITMQ_URL || 'amqp://admin:admin@localhost:5672'; - this.connection = amqp.connect([rabbitMQUrl]); + console.log('šŸ”Œ Connecting to RabbitMQ...'); + console.log(` URL: ${rabbitMQUrl.replace(/\/\/.*:.*@/, '//***:***@')}`); + + this.connection = amqp.connect([rabbitMQUrl], { + heartbeatIntervalInSeconds: 30, + reconnectTimeInSeconds: 5, + }); + + this.connection.on('connect', () => + console.log('āœ… RabbitMQ connection established'), + ); + this.connection.on('disconnect', (err) => + console.error( + 'āŒ RabbitMQ disconnected:', + // @ts-expect-error: RabbitMQ error type is unknown and may not contain "message" + err?.message || 'Unknown error', + ), + ); + + this.connection.on('connectFailed', (err) => + console.error( + 'āŒ RabbitMQ connection failed:', + // @ts-expect-error: RabbitMQ error type does not strictly match TypeScript's expected shape + err?.message || 'Unknown error', + ), + ); this.channelWrapper = this.connection.createChannel({ - json: true, + json: true, // IMPORTANT: Match API Gateway's json:true setting setup: async (channel: any) => { - await channel.assertQueue('email.queue', { durable: true }); + console.log('āš™ļø Setting up RabbitMQ channel...'); + + // Assert queues + await channel.assertQueue('email.queue', { + durable: true, + arguments: { + 'x-message-ttl': 86400000, // 24 hours + }, + }); await channel.assertQueue('failed.queue', { durable: true }); + console.log('šŸ“¬ Queues asserted'); + + // CHECK QUEUE STATUS BEFORE CONSUMING + const queueInfo = await channel.checkQueue('email.queue'); + console.log('šŸ“Š Queue Status BEFORE consuming:'); + console.log(` Messages in queue: ${queueInfo.messageCount}`); + console.log(` Consumers: ${queueInfo.consumerCount}`); + + // Set prefetch await channel.prefetch(1); - await channel.consume('email.queue', async (msg: any) => { - if (msg) { - await this.processEmailMessage(msg, channel); - } - }); + console.log('āš™ļø Prefetch set to 1'); + + // Start consuming + const consumerTag = await channel.consume( + 'email.queue', + async (msg: any) => { + if (msg) { + console.log('\n' + '='.repeat(60)); + console.log('šŸŽ‰ MESSAGE RECEIVED FROM QUEUE!'); + console.log('='.repeat(60)); + await this.processEmailMessage(msg, channel); + } else { + console.log('āš ļø Received null message'); + } + }, + { noAck: false }, + ); + + console.log(`āœ… Consumer started with tag: ${consumerTag.consumerTag}`); + + // CHECK QUEUE STATUS AFTER CONSUMING + const queueInfoAfter = await channel.checkQueue('email.queue'); + console.log('šŸ“Š Queue Status AFTER consumer setup:'); + console.log(` Messages in queue: ${queueInfoAfter.messageCount}`); + console.log(` Consumers: ${queueInfoAfter.consumerCount}`); + + console.log('\n' + 'šŸŽ§'.repeat(20)); + console.log('šŸ‘‚ EMAIL SERVICE IS NOW LISTENING FOR MESSAGES'); + console.log('šŸŽ§'.repeat(20) + '\n'); + + // If there are messages, they should be consumed immediately + if (queueInfoAfter.messageCount > 0) { + console.log( + `āš” ${queueInfoAfter.messageCount} message(s) waiting to be processed...`, + ); + } }, }); + this.channelWrapper.on('error', (err) => { + console.error('āŒ Channel error:', err.message); + }); + + this.channelWrapper.on('close', () => { + console.log('āš ļø Channel closed'); + }); + await this.channelWrapper.waitForConnect(); - console.log('Email Service connected to RabbitMQ'); + console.log('āœ… Email Service connected to RabbitMQ and ready'); } - private async processEmailMessage(msg: any, channel: any) { - const message = JSON.parse(msg.content.toString()); - const correlationId = message.notification_id; + private async processEmailMessage(msg: ConsumeMessage, channel: any) { + let correlationId = 'unknown'; try { - console.log(`Processing email for notification: ${correlationId}`); + // Parse message + const messageContent = msg.content.toString(); + console.log('šŸ“ Raw message content:', messageContent); + + const message = JSON.parse(messageContent); + correlationId = message.notification_id; + + console.log(`\n=== PROCESSING EMAIL ===`); + console.log(`Notification ID: ${correlationId}`); + console.log(`Recipient: ${message.user_email}`); + console.log(`Subject Template: ${message.template?.subject || 'N/A'}`); + + // Update status to PROCESSING await this.updateStatus(correlationId, NotificationStatus.PROCESSING); + // Compile templates const titleTemplate = Handlebars.compile( message.template.subject || 'Notification', ); const bodyTemplate = Handlebars.compile(message.template.content); - const subject = titleTemplate(message.variables); - const html = bodyTemplate(message.variables); + const subject = titleTemplate(message.variables || {}); + const html = bodyTemplate(message.variables || {}); + + console.log(`šŸ“§ Sending email with subject: "${subject}"`); // Send email const info = await this.transporter.sendMail({ @@ -114,21 +238,33 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { html: html, }); - console.log(`Email sent successfully: ${correlationId}`, info.messageId); + console.log(`āœ… Email sent successfully!`); + console.log(` Message ID: ${info.messageId}`); + console.log(` Response: ${info.response}`); // Update status to DELIVERED await this.updateStatus( correlationId, NotificationStatus.DELIVERED, null, - { smtp_message_id: info.messageId, recipient: message.user_email }, + { + smtp_message_id: info.messageId, + recipient: message.user_email, + sent_at: new Date().toISOString(), + }, ); // Acknowledge message channel.ack(msg); this.retryAttempts.delete(correlationId); + + console.log(`āœ… Message acknowledged and removed from queue`); + console.log(`=== EMAIL PROCESSING COMPLETE ===\n`); } catch (error) { - console.error(`Failed to send email: ${error.message}`); + console.error(`\nāŒ FAILED TO SEND EMAIL`); + console.error(` Notification ID: ${correlationId}`); + console.error(` Error: ${error.message}`); + console.error(` Stack: ${error.stack}`); const attempts = this.retryAttempts.get(correlationId) || 0; @@ -137,6 +273,8 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { this.retryAttempts.set(correlationId, attempts + 1); const delay = Math.pow(2, attempts) * 1000; + console.log(`šŸ”„ Will retry in ${delay}ms (attempt ${attempts + 1}/3)`); + // Update status to RETRYING await this.updateStatus( correlationId, @@ -145,17 +283,20 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { ); setTimeout(() => { + console.log(`šŸ”„ Requeuing message for retry...`); channel.nack(msg, false, true); }, delay); - - console.log( - `Retrying email (attempt ${attempts + 1}/3) after ${delay}ms`, - ); } else { // Move to dead letter queue - console.log(`Moving to dead letter queue: ${correlationId}`); - await channel.sendToQueue('failed.queue', Buffer.from(msg.content), { + console.log(`ā˜ ļø Max retries exceeded. Moving to dead letter queue.`); + + await channel.sendToQueue('failed.queue', msg.content, { persistent: true, + headers: { + 'x-original-queue': 'email.queue', + 'x-failed-at': new Date().toISOString(), + 'x-error': error.message, + }, }); // Update status to FAILED @@ -167,6 +308,7 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { channel.ack(msg); this.retryAttempts.delete(correlationId); + console.log(`=== EMAIL PROCESSING FAILED ===\n`); } } } @@ -178,13 +320,15 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { metadata?: any, ) { try { + // Update Redis const currentStatus = await this.redisClient.get( `status:${notificationId}`, ); + let updatedStatus; if (currentStatus) { const statusObj = JSON.parse(currentStatus); - const updatedStatus = { + updatedStatus = { ...statusObj, status: status, updated_at: new Date().toISOString(), @@ -194,18 +338,29 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { (statusObj.attempts || 0) + (status === NotificationStatus.RETRYING ? 1 : 0), }; + } else { + updatedStatus = { + notification_id: notificationId, + status: status, + updated_at: new Date().toISOString(), + error: error, + metadata: metadata, + attempts: status === NotificationStatus.RETRYING ? 1 : 0, + }; + } - await this.redisClient.setEx( - `status:${notificationId}`, - 86400, // 24 hours - JSON.stringify(updatedStatus), - ); + await this.redisClient.setEx( + `status:${notificationId}`, + 86400, // 24 hours + JSON.stringify(updatedStatus), + ); - console.log(`Status updated in Redis: ${notificationId} -> ${status}`); - } + console.log(`šŸ“Š Redis status updated: ${notificationId} -> ${status}`); + // Update via API Gateway const apiGatewayUrl = process.env.API_GATEWAY_URL || 'http://localhost:3000'; + try { await firstValueFrom( this.httpService.post( @@ -216,26 +371,47 @@ export class EmailService implements OnModuleInit, OnModuleDestroy { error: error, metadata: metadata, }, + { + timeout: 5000, // 5 second timeout + }, ), ); console.log( - `Status updated via API Gateway: ${notificationId} -> ${status}`, + `šŸ“” API Gateway status updated: ${notificationId} -> ${status}`, ); } catch (apiError) { - // Don't fail if API Gateway is unavailable - Redis update is sufficient + // Don't fail if API Gateway is unavailable console.warn( - 'Failed to update status via API Gateway (non-critical):', - apiError.message, + `āš ļø Failed to update status via API Gateway (non-critical): ${apiError.message}`, ); } } catch (err) { - console.error('Failed to update status:', err.message); + console.error(`āŒ Failed to update status: ${err.message}`); } } async onModuleDestroy() { - await this.channelWrapper.close(); - await this.connection.close(); - await this.redisClient.quit(); + console.log('šŸ›‘ Shutting down Email Service...'); + + try { + if (this.channelWrapper) { + await this.channelWrapper.close(); + console.log('āœ… Channel closed'); + } + + if (this.connection) { + await this.connection.close(); + console.log('āœ… RabbitMQ connection closed'); + } + + if (this.redisClient) { + await this.redisClient.quit(); + console.log('āœ… Redis connection closed'); + } + } catch (error) { + console.error('āŒ Error during shutdown:', error.message); + } + + console.log('šŸ‘‹ Email Service shut down complete'); } }