Skip to content

chore: install sendgrid #20

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
harjibbolar26 merged 1 commit into from
Nov 14, 2025
Merged

chore: install sendgrid #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 41 additions & 3 deletions email_service/package-lock.json
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions email_service/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
"@nestjs/passport": "^10.0.0",
"@nestjs/platform-express": "^10.0.0",
"@nestjs/swagger": "^7.0.0",
"@sendgrid/mail": "^8.1.6",
"amqp-connection-manager": "^4.1.14",
"amqplib": "^0.10.3",
"axios": "^1.6.0",
Expand Down
97 changes: 46 additions & 51 deletions email_service/src/email/email.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import { Injectable, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { HttpService } from '@nestjs/axios';
import { firstValueFrom } from 'rxjs';
import * as nodemailer from 'nodemailer';
import * as sgMail from '@sendgrid/mail';
import * as Handlebars from 'handlebars';
import * as amqp from 'amqp-connection-manager';
import { ChannelWrapper } from 'amqp-connection-manager';
Expand All @@ -20,21 +20,19 @@ enum NotificationStatus {
export class EmailService implements OnModuleInit, OnModuleDestroy {
private connection: amqp.AmqpConnectionManager;
private channelWrapper: ChannelWrapper;
private transporter: nodemailer.Transporter;
private retryAttempts = new Map<string, number>();
private redisClient: RedisClientType;
private isProcessing = false;

constructor(private readonly httpService: HttpService) {
this.transporter = nodemailer.createTransport({
host: process.env.SMTP_HOST || 'smtp.gmail.com',
port: parseInt(process.env.SMTP_PORT || '587'),
secure: false,
auth: {
user: process.env.SMTP_USER,
pass: process.env.SMTP_PASS,
},
});
// Initialize SendGrid
const apiKey = process.env.SENDGRID_API_KEY;
if (!apiKey) {
console.error('❌ SENDGRID_API_KEY environment variable is not set');
} else {
sgMail.setApiKey(apiKey);
console.log('✅ SendGrid initialized');
}
}

async onModuleInit() {
Expand All @@ -46,18 +44,25 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
// Connect to RabbitMQ
await this.connectRabbitMQ();

// Verify SMTP connection
await this.verifySmtpConnection();
// Verify SendGrid
await this.verifySendGrid();
}

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 verifySendGrid() {
if (!process.env.SENDGRID_API_KEY) {
console.error('❌ SendGrid API key not configured');
console.error('Set SENDGRID_API_KEY environment variable');
return;
}

if (!process.env.FROM_EMAIL) {
console.error('❌ FROM_EMAIL not configured');
console.error('Set FROM_EMAIL to your verified sender email');
return;
}

console.log('✅ SendGrid ready to send emails');
console.log(` From: ${process.env.FROM_EMAIL}`);
}

private async connectRedis() {
Expand Down Expand Up @@ -126,32 +131,28 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
);

this.channelWrapper = this.connection.createChannel({
json: true, // IMPORTANT: Match API Gateway's json:true setting
json: true,
setup: async (channel: any) => {
console.log('⚙️ Setting up RabbitMQ channel...');

// Assert queues
await channel.assertQueue('email.queue', {
durable: true,
arguments: {
'x-message-ttl': 86400000, // 24 hours
'x-message-ttl': 86400000,
},
});
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);
console.log('⚙️ Prefetch set to 1');

// Start consuming
const consumerTag = await channel.consume(
'email.queue',
async (msg: any) => {
Expand All @@ -169,7 +170,6 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {

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}`);
Expand All @@ -179,7 +179,6 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
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...`,
Expand All @@ -204,7 +203,6 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
let correlationId = 'unknown';

try {
// Parse message
const messageContent = msg.content.toString();
console.log('📝 Raw message content:', messageContent);

Expand All @@ -216,10 +214,8 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
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',
);
Expand All @@ -230,31 +226,32 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {

console.log(`📧 Sending email with subject: "${subject}"`);

// Send email
const info = await this.transporter.sendMail({
from: process.env.FROM_EMAIL || process.env.SMTP_USER,
// Send email via SendGrid
const emailMsg = {
to: message.user_email,
from: process.env.FROM_EMAIL,
subject: subject,
html: html,
});
};

const response = await sgMail.send(emailMsg);

console.log(`✅ Email sent successfully!`);
console.log(` Message ID: ${info.messageId}`);
console.log(` Response: ${info.response}`);
console.log(`✅ Email sent successfully via SendGrid!`);
console.log(` Status Code: ${response[0].statusCode}`);
console.log(` Message ID: ${response[0].headers['x-message-id']}`);

// Update status to DELIVERED
await this.updateStatus(
correlationId,
NotificationStatus.DELIVERED,
null,
{
smtp_message_id: info.messageId,
sendgrid_message_id: response[0].headers['x-message-id'],
recipient: message.user_email,
sent_at: new Date().toISOString(),
status_code: response[0].statusCode,
},
);

// Acknowledge message
channel.ack(msg);
this.retryAttempts.delete(correlationId);

Expand All @@ -264,18 +261,21 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
console.error(`\n❌ FAILED TO SEND EMAIL`);
console.error(` Notification ID: ${correlationId}`);
console.error(` Error: ${error.message}`);
console.error(` Stack: ${error.stack}`);

// SendGrid specific error details
if (error.response) {
console.error(` Status: ${error.code}`);
console.error(` Body: ${JSON.stringify(error.response.body)}`);
}

const attempts = this.retryAttempts.get(correlationId) || 0;

if (attempts < 3) {
// Retry with exponential backoff
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,
NotificationStatus.RETRYING,
Expand All @@ -287,7 +287,6 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
channel.nack(msg, false, true);
}, delay);
} else {
// Move to dead letter queue
console.log(`☠️ Max retries exceeded. Moving to dead letter queue.`);

await channel.sendToQueue('failed.queue', msg.content, {
Expand All @@ -299,7 +298,6 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
},
});

// Update status to FAILED
await this.updateStatus(
correlationId,
NotificationStatus.FAILED,
Expand All @@ -320,7 +318,6 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
metadata?: any,
) {
try {
// Update Redis
const currentStatus = await this.redisClient.get(
`status:${notificationId}`,
);
Expand Down Expand Up @@ -351,13 +348,12 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {

await this.redisClient.setEx(
`status:${notificationId}`,
86400, // 24 hours
86400,
JSON.stringify(updatedStatus),
);

console.log(`📊 Redis status updated: ${notificationId} -> ${status}`);

// Update via API Gateway
const apiGatewayUrl =
process.env.API_GATEWAY_URL || 'http://localhost:3000';

Expand All @@ -372,15 +368,14 @@ export class EmailService implements OnModuleInit, OnModuleDestroy {
metadata: metadata,
},
{
timeout: 5000, // 5 second timeout
timeout: 5000,
},
),
);
console.log(
`📡 API Gateway status updated: ${notificationId} -> ${status}`,
);
} catch (apiError) {
// Don't fail if API Gateway is unavailable
console.warn(
`⚠️ Failed to update status via API Gateway (non-critical): ${apiError.message}`,
);
Expand Down