DEV Community: Bhaskar Sawant

How to Set Up Scalable Queue Workers on AWS Using ElastiCache, ECS, and BullMQ

Bhaskar Sawant — Sun, 13 Oct 2024 14:10:18 +0000

BullMQ is a powerful and flexible job queuing system that helps manage background tasks in Node.js applications. As your application grows, it’s crucial to ensure the scalability and resilience of your job processing infrastructure.

In this guide, you’ll learn how to set up BullMQ workers on AWS using ElastiCache for Redis as your data store and ECS (Elastic Container Service) for deploying and managing the queue workers.

What You Will Learn

Setting Up BullMQ Queues and Workers Locally

Before deploying to AWS, let’s start with setting up a BullMQ worker in your local environment.

1. Start a Redis Container in Docker

To run Redis locally, use Docker:

docker run --name redis-container -p 6379:6379 -d redis

This command pulls the latest Redis image from Docker Hub (if not already available locally) and starts a Redis container, exposing the port 6379 for communication with BullMQ.

2. Initialize a Node.js Project

Next, let’s set up your Node.js project. Navigate to your desired project directory and initialize the project using:

npm init -y

For this tutorial, we’ll follow this simple project structure:

BullMQ/
├── src/
│   ├── worker.js
├── .env
├── dockerfile
└── package.json

3. Install Required Packages

Now add two packages,

npm i bullmq dotenv

4. Configure the Worker

Create a file worker.js inside the src/ directory.

This file will contain the code for setting up the BullMQ queue and processing jobs. Start by importing the necessary modules and configuring Redis:

// worker.js
import dotenv from "dotenv"
import { Queue, Worker } from "bullmq";
dotenv.config()

const connection = {
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT, 
};


// Initialize the queue
const myQueue = new Queue('myQueue', { connection });

Next, create a .env file in your root directory with the following configuration:

REDIS_HOST=localhost
REDIS_PORT=6379

This connects BullMQ to the local Redis instance running on localhost:6379.

5. Add a Demo Job to the Queue

To demonstrate how the queue works, we’ll add a demo job if the queue is empty.

Add the following code to worker.js:

// worker.js

// Add a job to the queue if it's empty (for demo purposes)
myQueue.getJobCounts().then((counts) => {
    if (counts.waiting === 0 && counts.active === 0) {
        console.log('Queue is empty, adding a demo job.');
        myQueue.add('myJob', { foo: 'bar' });
    }
});

This snippet checks the queue for unprocessed jobs. If no jobs are waiting or active, it adds a demo job with some sample data ({ foo: 'bar' }).

6. Add a Worker to Process Jobs

Now, let’s create a worker to process jobs from the queue:

// worker.js

// Worker logic to process jobs
const worker = new Worker('myQueue', async (job) => {

    console.log(`Processing job ${job.id} with data:`, job.data);

    await new Promise((resolve) => setTimeout(resolve, 1000)); // Simulate job processing delay

    console.log(`Job ${job.id} processed successfully`);

}, { connection });

// Error handling for failed jobs
worker.on('failed', (job, err) => {
    console.error(`Job ${job.id} failed with error: ${err.message}`);
});

// Event listener for job completion
worker.on('completed', (job) => {
    console.log(`Job ${job.id} completed successfully`);
});

The worker processes each job from the queue, logging the job’s id and data. It also handles job failures and logs successful completions. For demonstration, the job takes 1 second to process.

7. Continuously Add Jobs for Testing

To simulate a constant stream of jobs, add a new job every second:

// worker.js

let count = 0

setInterval(() => {
    count++
    myQueue.add('myJob', { foo: count });
    console.log('adding a demo job.', count);
}, 1000)

This will automatically add a new job to the queue every second, helping you test the worker’s ability to process multiple jobs.

8. Run the Project

Finally, add a script in your package.json to run the worker:

  "scripts": {
    "start": "node src/worker.js"
  }

Now, you can run the project using the following command:

npm run start

This will start the worker, connect it to Redis, and continuously add and process jobs.

Creating Redis database using ElastiCache

To ensure that your BullMQ workers can communicate seamlessly with Redis on AWS, we’ll set up a Redis database using Amazon ElastiCache.

ElastiCache is a fully managed in-memory data store that supports Redis, enabling high-performance and scalable job processing.

1. Create a Security Group for Redis and ECS Communication

Before setting up Redis, we need to create a security group that will allow communication between your ElastiCache Redis instance and your ECS workers.

1. Search for Security Groups: In the AWS console, use the search bar to find Security Groups.

2. Create a New Security Group:

Click on the Create Security Group button.
Name the Security Group: For example, you can name it

BullMQ-Worker.
Select Your VPC: Ensure you choose the correct VPC where your ECS tasks and ElastiCache will reside.

3. Configure Inbound Rules:

Add inbound rules to allow your ECS tasks to communicate with Redis. For instance, add a rule to allow traffic on port 6379 (the default Redis port).
Keep outbound rules as they are by default.

4. Create the Security Group: After configuring the settings, click on the Create Security Group button.

2. Set Up the Redis Database Using ElastiCache

Now that the security group has been created, let’s proceed to set up the Redis database on ElastiCache.

1. Search for ElastiCache: In the AWS console search bar, type ElastiCache and select it from the results.

2. Select Redis OSS Caches: On the ElastiCache dashboard, choose Redis from the left sidebar under the Redis OSS caches option.

3. Create a New Redis Cache: Click on the Create Redis OSS Cache button.

4. Configuration Settings:

Deployment Option: Select Design your own cache.
Creation Method: Choose Cluster cache.
Cluster Mode: Set this to Disabled (for single-node Redis setups).

5. Cluster Info:

Cluster Name: Provide a name for your Redis cluster, such as redisCluster.

6. Cluster Settings:

Node Type: For demonstration purposes, you can select a cost-effective instance like t2.micro.
Number of Replicas: Choose 1 replica to ensure redundancy.

7. Subnet Group Settings:

Subnet Group: If you have a default subnet group, select it. Otherwise, you may need to create one.
To create a new subnet group, go to the left sidebar under Configuration, select Subnet Groups, and create a group with the default settings.

8. Next Steps: After configuring the above settings, click on the Next button.

3. Configure Security and Access Control

1. Encryption in Transit: Enable Encryption in transit to secure communication between your Redis instance and workers.

2. Access Control (AUTH):

Enable AUTH Default User Access. This requires you to set a password, also known as an auth token.
Set Password: Enter a password in the Auth Token input box. This password will be used to authenticate connections to Redis from your BullMQ worker.

3. Select Security Group:

In the Selected Security Groups section, select the security group you created earlier (BullMQ-Worker).

4. Other Settings: Leave other settings as they are by default.

4. Review and Create the Redis Cache

Review Configurations: Review your configurations on the final page to ensure all the settings are correct.
Create Redis Cache: Once you are satisfied with the settings, click on Create to set up your Redis cache. It will take some time to create your Redis database.

5. Add a new configuration in the worker

since we have shifted our Redis from local to ElastiCache we need to pass a new configuration to the worker.

// worker.js

const connection = {
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT,
    password: process.env.REDIS_PASSWORD,
    tls: { rejectUnauthorized: false }, 
};

here we have added password and tls because we added encryption in transit.

Building a Docker Container for Your BullMQ Worker

The first step in deploying your BullMQ worker on AWS is containerising the application using Docker.

A Docker container ensures that your worker will run consistently across different environments, making it ideal for cloud deployments.

Let’s create a Dockerfile to build the container.

1. Add a Dockerfile to Your Project

In the root directory of your project, create a new file called Dockerfile and add the following content:


# dockerfile

FROM node:18

WORKDIR /

COPY package*.json .

RUN npm install

COPY src/ src/

CMD [ "npm", "start" ]

2. How the Dockerfile Works

1. Base Image: The node:18 image is a version of Node.js, ensuring your container is optimized in terms of size and speed.

2. Working Directory: The WORKDIR /app command sets the working directory inside the container where your application will reside.

3. Copy Dependencies:

The COPY package*.json ./ command copies your package.json and package-lock.json files into the container.

4. Install Dependencies: The RUN npm installcommand installs the necessary packages.

5. Copy Source Code: The COPY src/ ./src command copies your source files (such as worker.js) into the container.

6. Run the Worker: The CMD ["npm", "start"] command runs the worker using the npm start script, as defined in your package.json.

This setup ensures that your worker runs efficiently in a Docker container, providing consistency whether you’re running locally or in production on AWS.

Pushing your Docker container to Amazon ECR

To deploy your BullMQ worker on Amazon ECS, you first need to push your Docker container to the Amazon Elastic Container Registry (ECR).

ECR is a fully managed Docker container registry that makes it easy to store, manage, and deploy container images.

1. Create a Repository on Amazon ECR

Login to AWS Console: Start by logging into your AWS Console.
Search for ECR: In the search bar, type “ECR” and select Amazon Elastic Container Registry from the search results.
Create a New Repository:

On the ECR dashboard, click the Create Repository button.
Enter a name for your repository. For example, you can name it bullmq-worker or any other name that reflects your project.
Leave the default settings or customize them as needed, then hit the Create button.

2. Push Your Docker Container to ECR

Once your repository is created, you’ll need to push your Docker image to ECR. Follow these steps:

Select Your Repository: On the ECR dashboard, find and click on the repository you just created.
View Push Commands: Inside your repository, click on the View push commands button.
Execute Push Commands: Follow the instructions provided by AWS ECR to push your Docker container and make sure to execute these commands in the dir where you have your dockerfile.

Deploying your Queue workers on AWS using ECS

After pushing your Docker container to Amazon ECR and setting up your Redis database using ElastiCache, the final step is deploying your BullMQ workers on Amazon ECS.

ECS (Elastic Container Service) enables you to run and manage your Docker containers efficiently in the cloud.

Let’s go through the process of deploying your workers.

1. Create a Task Definition for Your Worker

1. Search for ECS: In the AWS console, type ECS in the search bar and navigate to the ECS dashboard.

2. Select Task Definitions: From the left sidebar, click on Task Definitions.

3. Create a New Task Definition:

Click the Create new task definition button.
Task Definition Family Name: Enter a name for your task definition, for example, bull-worker-tasks.

4. Infrastructure Requirements:

In the Operating System/Architecture section:
Choose Linux/x86_64 if your Docker container was built on Intel or AMD-based machines.
Select Linux/ARM64 if your container was built on Apple M1/M2 chips or ARM-based processors.

5. Container Details:

Container Name: Give your container a name, such as bull-worker.
Image URI: In the Image URI input field, paste the URI of the repository from Amazon ECR, which you created earlier.

6. Port Mappings:

Add a port mapping for port 6379 (Redis’s default port).
Set the App Protocol to HTTP.

7. Environment Variables:

Add the following environment variables:
REDIS_HOST: This can be found in the details of your ElastiCache Redis database. Go to the ElastiCache dashboard, select your cluster, and copy the Primary Endpoint.
REDIS_PORT: Set this to 6379.
REDIS_PASSWORD: This is the password you set when creating your Redis database on ElastiCache.

8. Create the Task Definition: After filling in the details, click the Create button.

2. Create an ECS Cluster

1. Navigate to Clusters: In the ECS dashboard, select Clusters from the left sidebar.

2. Create a New Cluster:

Click the Create cluster button.
Enter a name for your cluster, such as bullMQ-worker-cluster.
Hit Create to finalize the cluster creation.

3. Select Your Cluster: Go back to the clusters homepage and select the cluster you just created.

3. Create and Configure an ECS Service

1. Create a Service: In your cluster’s detail page, scroll down to the Services section and click Create Service.

2. Deployment Configuration:

Scroll down to the Deployment Configuration section.
Under Task Definition Family, select the task definition you created earlier, e.g., bull-worker-tasks.
Give your service a unique name, such as bullMQ-worker-service.

3. Networking: In the Networking section, select the security group you created for the ElastiCache cluster (bullMQ-group).

4. Create the Service: Once everything is configured, hit the Create button.

4. Monitor Your ECS Cluster and Worker Logs

Wait for Tasks to Start: Return to your ECS cluster’s detail page and wait until your task starts running. ECS will automatically pull the Docker image from ECR and spin up your BullMQ worker.
View Logs:

After the task is running, go to the Services section, and click on the name of your service.

On the service details page, click on the Logs tab to monitor the logs from your BullMQ worker. Here, you’ll be able to see logs for the queue jobs being processed.

Scaling Your Queue Workers and Redis Database on AWS

Now that we’ve deployed BullMQ queue workers and a Redis database on AWS, it’s crucial to understand how to scale them to handle growing workloads efficiently.

With Amazon ECS and ElastiCache, you can scale both horizontally (adding more workers/nodes) and vertically (upgrading instance sizes).

1. Scaling Your Redis Database

Amazon ElastiCache allows you to scale Redis horizontally (adding more nodes or replicas) or vertically (upgrading to larger instance types).

Vertical Scaling for Redis

Vertical scaling of Redis involves upgrading the node type to a larger instance to handle more load. Here’s how you can do it:

Go to ElastiCache Dashboard: From your AWS console, search for ElastiCache.
Select Your Redis Cluster: In the Redis section, select the Redis cluster you want to scale.
Modify Cluster:

Click on Modify at the top of the cluster details page.
Under Node Type, select a larger instance size (e.g., from t2.micro to r5.large).

Apply Changes: After making the selection, choose whether to apply the changes during the next maintenance window or immediately.

Vertical scaling is best suited for situations where Redis is CPU-bound or memory-bound due to the large number of jobs being processed.

Horizontal Scaling for Redis

Horizontal scaling involves adding more Redis replicas to increase availability and distribute the load.

Go to ElastiCache Dashboard: Open the Redis section in the ElastiCache dashboard.
Select Your Redis Cluster and scroll down to the services section there you’ll find the Add node button click on it then it will open a popup to ask the number of replicas you want to create.
Each new replica will serve read requests, reducing the load on the primary node.
Apply Changes: You can apply these changes.

2. Scaling Your ECS Queue Workers

Scaling the ECS workers horizontally allows you to add more workers that process jobs concurrently, while vertical scaling involves increasing the power of individual ECS instances.

Horizontal Scaling for ECS Workers

To scale your queue workers horizontally, you can add more tasks to your ECS service, enabling multiple instances of your worker container to run in parallel.

Go to ECS Console: In the AWS console, search for ECS and navigate to the cluster that you’ve set up for BullMQ workers.
Select Your Service: Go to the Services tab and select the BullMQ worker service.
Update Desired Task Count:

Click Update Service.
In the Desired Task field, increase the number of tasks (e.g., from 1 to 3).
Apply Changes: Click Update to scale your service.

ECS will automatically spin up more instances of your worker to handle an increased number of jobs in the queue. Horizontal scaling is particularly useful when the queue length increases, as additional workers can process jobs simultaneously.

Vertical Scaling for ECS Workers

Since we have selected the AWS Fargate(serverless) we don’t need to worry about manually scaling the underlying infrastructure (e.g., EC2 instances) because Fargate automatically provisions the required resources for your tasks.

But if you want more control of your infrastructure and you’ve selected the Amazon EC2 instances, then To scale vertically, upgrade the ECS instance type to handle more powerful tasks.

Go to ECS Console and select Clusters.
Select Cluster Capacity Provider: Choose the Capacity Provider associated with your cluster.
Modify EC2 Instance Type:

In the Auto Scaling configuration, you can change the instance type (e.g., from t3.medium to m5.large).
Update Service: Apply changes to your ECS service for the updated instance type to take effect.

Vertical scaling is best used when your worker instances need more CPU or memory resources to process individual jobs faster.

Bonus:

In this article, we’ve explored how to set up a scalable queue worker infrastructure on AWS using BullMQ. Exciting, right? But here’s the thing — if you noticed, we manually scaled the workers and Redis instances. So, you might be wondering: what about auto-scaling? Don’t worry, I’ve got you covered!

In my next article, I’ll dive deep into how to implement auto-scaling for your queue workers on AWS, making your infrastructure truly hands-off and ready to handle spikes in traffic automatically. Whether you’re a beginner eager to learn or a seasoned pro looking to fine-tune your setup, this guide will have something for everyone.

So, if you want to master auto-scaling with BullMQ and AWS, be sure to subscribe and follow me on LinkedIn — you won’t want to miss this!

Stay tuned for the next chapter in building scalable architectures with BullMQ!

Follow Me for More Insights

If you found this guide helpful and want to learn more about web development, Software Engineering, and other exciting tech topics, follow me here on Medium. I regularly share tutorials, insights, and tips to help you become a better developer. Let’s connect and grow together in this journey of continuous learning!

Connect with Me

Twitter: https://twitter.com/Bhaskar_Sawant_
LinkedIn: https://www.linkedin.com/in/bhaskar-sawant/
GitHub: https://github.com/bhaskarcsawant

Stay tuned for more articles, and feel free to reach out if you have any questions or suggestions for future topics.

Cheers, thanks for reading! 😊

Happy coding! 🚀

This article was originally published on Medium. It is also available on Hashnode and Dev.to.

How to Build Scalable Automated Workflows like HubSpot using AWS Lambda, SQS, Node.js, and MongoDB

Bhaskar Sawant — Sun, 04 Aug 2024 07:19:33 +0000

Creating automated workflows similar to HubSpot’s powerful system involves leveraging advanced cloud technologies. This guide will walk you through the process of developing such workflows using AWS Lambda, Amazon SQS, Node.js, and MongoDB.

Understanding Automated Workflows

Automated workflows are sequences of actions triggered by specific events. They streamline repetitive tasks, send notifications, update records, and more, all without manual intervention. By integrating AWS Lambda, SQS, Node.js, and MongoDB, you can build robust workflows. This tutorial provides a step-by-step approach to creating these workflows.

What You Will Learn

Automated Workflow Architecture: Gain insights into the inner workings of automated workflows.
Understanding API Server: Learn how workflows operate through code examples and explanations.
Configure AWS CLI: Install and configure the AWS CLI to manage AWS resources from the command line.
Deploy with Serverless Framework: Create and deploy AWS Lambda functions and other resources using the Serverless Framework.
Integrate API with AWS SQS: Send messages to AWS SQS from your API server for reliable workflows.
Deploy API Server on AWS EC2: Deploy your Node.js API server to AWS EC2 for high availability and performance.
Scalability: Understand how to scale your workflows and API server to handle increased traffic and demand.

Prerequisites

An AWS account.
Basic knowledge of Node.js and MongoDB.
Access to a MongoDB Atlas account or a local MongoDB setup.

Automated Workflow Architecture.

Let’s illustrate this with an example where a workflow sends a message to a user upon a form submission. Here’s the step-by-step process:

Trigger Event: When a form submission occurs, it acts as a trigger.
Send Message to SQS: Add a piece of code in the form submission function to push a message to an SQS queue named TriggerQueue on every new form submission.
Invoke Lambda Function: TriggerQueue triggers a Lambda function upon receiving a message.
Check for Workflow in DB: The Lambda function checks if any workflow exists in the database that matches the configuration provided in the SQS message.
Fetch Workflow Actions: If a matching workflow is found, it fetches the set of actions to perform with the provided data.
Send Actions to Second SQS Queue: The Lambda function sends the actions to another SQS queue named ActionsQueue along with the trigger data.
Invoke Second Lambda Function: ActionsQueue triggers a second Lambda function upon receiving a message.
Perform Actions: The second Lambda function takes the action data and calls the respective APIs to perform the actions, such as sending a message to the user.

This setup ensures that your workflows are reliable and can handle various tasks efficiently.

Understanding API Server.

You can find the source code for the project here.

Now, let’s dive into the Model and Controller to understand how the API server functions.

Workflow Model

In a workflow, we have one Trigger and one or more Actions. Below is the schema definition for the workflow and trigger

// models/workflow.ts

const triggerSchema = new Schema<Trigger>({
  type: { type: String, required: true },
  app: { type: String, required: true },
  metaData: { type: String , required: false },
  data: { type: Schema.Types.Mixed, required: false } // Mixed type for dynamic data
});

const workflowSchema = new Schema<Workflow>({
  workflowName: { type: String, required: true },
  trigger: { type: triggerSchema, required: true },
  actions: { type: [triggerSchema], required: true },
  created_at: { type: Date, default: Date.now },
  updated_at: { type: Date, default: Date.now },
});

// Create and export the model
const WorkflowModel = mongoose.model<Workflow>('Workflow', workflowSchema);

export default WorkflowModel;

In this model:

Trigger: Contains properties such as type, app, optional metaData, and data.
Workflow: Extends the Mongoose Document interface, including properties like workflowName, trigger, actions, created_at, and updated_at.

For example, if our trigger is a form submission, the app will be 'forms' and the type will be 'new_submission'. The metaData helps to identify specific workflows, like so:

{
 "app": "forms",
 "type": "new_submission",
 "metadata":"{_id:someUniqueId}"
}

Creating a Form Submission

Here’s how you can handle a form submission and trigger workflows:

export const createFormSubmission = async (req: Request<{}, {}, FormSubmissionBody>, res: Response): Promise<void> => {
  try {
    const { formId, customerName, customerEmail, customerPhone } = req.body;
    const newForm = new FormSubmission({ formId, customerName, customerEmail, customerPhone });
    await newForm.save();

    // Prepare the data to be sent to the workflow
    let workflowData = {
      app: 'forms',
      type: 'new_submission',
      metaData: JSON.stringify({ _id: formId }),
      data: newForm,
    }

    let params: params = {
      MessageBody: JSON.stringify(workflowData),
      QueueUrl: process.env.AWS_SQS_URL as string,
    };

    // Send the message
    sqs.sendMessage(params, (err, data) => {
      if (err) {
        console.log('Error', err);
      } else {
        console.log('Success', data.MessageId);
      }
    });
    res.status(201).json(newForm);
  } catch (error) {
    res.status(500).json({ error: 'Internal server error' });
  }
};

In this controller:

A new form submission is created and saved to the database.
Workflow data is prepared, including the form submission details.
The workflow data is sent to an AWS SQS queue.
Any workflows matching the trigger configuration (e.g., app: 'forms', type: 'new_submission', metaData: { _id: someUniqueId}) will be identified and the corresponding actions will be sent to another queue to execute.

Configure AWS CLI.

To use the Serverless Framework, we first need to configure the AWS CLI. Follow these steps to create an IAM user and configure the AWS CLI.

Create an IAM User

Sign in to AWS Console: Log in to your AWS Management Console.
Search for IAM: In the search bar, type “IAM” and select the IAM service from the results.
Navigate to Users: Click on the “Users” option in the IAM dashboard.
Create a New User:

Click on “Add User.”
Enter a username.
Click “Next.”

Set Permissions:

Select “Attach policies directly.”
In the “Permission policies” section, select “AdministratorAccess.”
Click “Next” and then “Create User.”

Create Access Key:

Click on the newly created user.
Go to the “Security credentials” tab.
Click on “Create access key.”
Select “Command Line Interface (CLI)” as the use case.
Check the acknowledgment checkbox.
Click “Next.”

Add Description and Create Key:

Add a description if desired.
Click “Create Access Key.”

Store Access Keys:

Copy the Access Key ID and Secret Access Key.
Store them securely, as the Secret Access Key will not be shown again.

Configure AWS CLI

Install AWS CLI: If you haven’t already, download and install AWS CLI or brew install awscli .
Open Terminal/Command Prompt: Open your terminal (Linux/Mac) or command prompt (Windows).
Run AWS Configure Command:

aws configure

4. Enter Credentials: You’ll be prompted to enter your credentials.

AWS Access Key ID: Enter your Access Key ID.
AWS Secret Access Key: Enter your Secret Access Key.
Default region name: Enter your preferred AWS region (e.g., us-east-1).
Default output format: Enter your preferred output format (e.g., json).

5. Verify Configuration:

Run the following command to verify the configuration:

aws sts get-caller-identity

This should return details about your IAM user, confirming that the AWS CLI is configured correctly.

Deploy with Serverless Framework.

You can find the source code for the project here.

What is the Serverless Framework?

The Serverless Framework is an open-source tool that simplifies the process of building and deploying serverless applications. It abstracts away the complexities of cloud infrastructure, enabling developers to focus on writing application code. Supporting various cloud providers such as AWS, Azure, and Google Cloud, the Serverless Framework offers a consistent experience for deploying functions, managing resources, and scaling applications effortlessly.

In this guide, we will develop Lambda functions and deploy resources using the Serverless Framework. We will cover topics such as database calls from Lambda, using MongoDB, and pushing messages from one Lambda queue worker to another queue.

Install Serverless Framework: First, install the Serverless Framework globally on your machine.

npm install -g serverless

2. Create a serverless service: Navigate to your project directory and create a new Serverless service.

Enter serverless command in your terminal, and select AWS / Node.js / Simple Function option.
Now name your project, select the Create A New App option and give an app name, and this will create a serverless service.

3. Create Workers/handlers

TriggerQueue Handler: This handler will pull messages from the trigger queue, find workflows with the same trigger configuration provided in the message, and push the respective set of actions into the actions queue along with the trigger data.

'use strict';

const { MongoClient } = require('mongodb');
const AWS = require('aws-sdk');
const sqs = new AWS.SQS();

const MONGODB_URI = process.env.MONGODB_URI;
const MONGODB_DB = process.env.MONGODB_DB;
const OUTPUT_QUEUE_URL = process.env.OUTPUT_QUEUE_URL;

let cachedDb = null;

async function connectToDatabase(uri) {
  if (cachedDb) {
    return cachedDb;
  }

  const client = await MongoClient.connect(uri, {
    useNewUrlParser: true,
    useUnifiedTopology: true,
  });

  const db = client.db(MONGODB_DB);
  console.log('Connected to MongoDB');

  cachedDb = db;
  return db;
}

const buildQuery = (type, app, metaData) => {
  const query = { 'trigger.type': type };
  if (app) {
    query['trigger.app'] = app;
  }
  if (metaData) {
    query['trigger.metaData'] = metaData;
  }
  return query;
};

module.exports.triggerHandler = async (event) => {
  // Validate environment variables
  if (!MONGODB_URI || !MONGODB_DB || !OUTPUT_QUEUE_URL) {
    console.error('Missing environment variables');
    return {
      statusCode: 500,
      body: JSON.stringify({
        message: 'Internal Server Error',
      }),
    };
  }

  let db;
  try {
    db = await connectToDatabase(MONGODB_URI);
  } catch (error) {
    console.error('Error connecting to MongoDB:', error);
    return {
      statusCode: 500,
      body: JSON.stringify({
        message: 'Error connecting to MongoDB',
      }),
    };
  }

  for (const record of event.Records) {
    const { body } = record;
    console.log(`Received message: ${body}`);

    let parsedBody;
    try {
      parsedBody = JSON.parse(body);
    } catch (error) {
      console.error('Error parsing message body:', error);
      continue; // Skip to the next message
    }

    const { type, app, metaData, data } = parsedBody;

    let result;
    try {
      const collection = db.collection('workflows');
      const query = buildQuery(type, app, metaData);
      result = await collection.find(query).toArray();
    } catch (error) {
      console.error('Error querying MongoDB:', error);
      continue; // Skip to the next message
    }

    for (const workflow of result) {
      const MessageBody = {
        actions: workflow.actions,
        data: data,
      };
      console.log('Message to send:', MessageBody);

      const params = {
        QueueUrl: OUTPUT_QUEUE_URL,
        MessageBody: JSON.stringify(MessageBody),
      };

      try {
        await sqs.sendMessage(params).promise();
        console.log('Message sent to output queue');
      } catch (error) {
        console.error('Error sending message to output queue:', error);
      }
    }
  }

  return {
    statusCode: 200,
    body: JSON.stringify({
      message: 'Messages processed and sent successfully',
    }),
  };
};

ActionsQueue Handler: This handler pulls messages from the actions queue and calls the respective APIs to perform the workflow actions.

'use strict';
const axios = require('axios');


const API_SERVER_URL = process.env.API_SERVER_URL;


const handleMessage = async (type, data) => {
  try {
    switch (type) {
      case 'send_message':
        const messageBody = {
          from: 'workflow',
          message: JSON.stringify(data),
        };
        await axios.post(`${API_SERVER_URL}/api/v1/send_message`, messageBody);
        console.log('Message sent successfully');
        break;
      default:
        console.log('Unknown type:', type);
    }
  } catch (error) {
    console.error('Error handling message:', error);
  }
};

const performAction = async (action, data) => {
  const { type, app } = action;
  switch (app) {
    case 'messaging':
      await handleMessage(type, data);
      break;
    default:
      console.log('Unknown app:', app);
  }
};

module.exports.actionsHandler = async (event) => {
  for (const record of event.Records) {
    const { body } = record;

    let parsedBody;
    try {
      parsedBody = JSON.parse(body);
    } catch (error) {
      console.error('Error parsing message body:', error);
      continue; // Skip to the next record
    }

    const { actions, data } = parsedBody;
    for (const action of actions) {
      console.log('Performing action:', action, data);
      try {
        await performAction(action, data);
        console.log('Action performed successfully');
      } catch (error) {
        console.error('Error performing action:', error);
      }
    }
  }

  return {
    statusCode: 200,
    body: JSON.stringify({
      message: 'Messages processed and sent successfully',
    }),
  };
};

4. Deploy the Serverless Service: The serverless.yml configuration file outlines the setup for deploying a serverless application using the Serverless Framework on AWS.

Here's a breakdown in simple terms:

Basic Information:

Organization: bhaskarsawant
Application: trigger
Service: trigger

Plugins:

serverless-offline: Allows running the service locally for development and testing.

Provider Configuration:

Provider: AWS
Runtime: Node.js version 20.x
Region: ap-south-1 (AWS region)

IAM Role Statements

Permissions: The service is allowed to:
Receive messages from SQS
Delete messages from SQS
Get SQS queue attributes
Send messages to SQS

Resources:

TriggerQueue ARN (Amazon Resource Name)
ActionsQueue ARN

Environment Variables:

OUTPUT_QUEUE_URL: Reference to the ActionsQueue
MONGODB_URI: MongoDB connection URL (from environment variables)
MONGODB_DB: MongoDB database name (from environment variables)
API_SERVER_URL: API server URL (from environment variables)

Functions:

sqsHandler:
Handler: handler.triggerHandler
Trigger: SQS queue TriggerQueue with a batch size of 1
actionsHandler:
Handler: actionsHandler.actionsHandler
Trigger: SQS queue ActionsQueue with a batch size of 1

Resources

TriggerQueue:
Type: AWS SQS Queue
Name: trigger_queue
ActionsQueue:
Type: AWS SQS Queue
Name: actions_queue

org: bhaskarsawant
app: trigger
service: trigger

plugins:
  - serverless-offline

provider:
  name: aws
  runtime: nodejs20.x
  region: ap-south-1
  iamRoleStatements:
    - Effect: "Allow"
      Action:
        - "sqs:ReceiveMessage"
        - "sqs:DeleteMessage"
        - "sqs:GetQueueAttributes"
        - "sqs:SendMessage"
      Resource:
        - !GetAtt TriggerQueue.Arn
        - !GetAtt ActionsQueue.Arn
  environment:
    OUTPUT_QUEUE_URL:
      Ref: ActionsQueue
    MONGODB_URI: ${env:DB_URL}
    MONGODB_DB: ${env:DB_NAME}
    API_SERVER_URL: ${env:API_URL}

functions:
  sqsHandler:
    handler: handler.triggerHandler
    events:
      - sqs:
          arn:
            Fn::GetAtt:
              - TriggerQueue
              - Arn
          batchSize: 1
  actionsHandler:
    handler: actionsHandler.actionsHandler
    events:
      - sqs:
          arn:
            Fn::GetAtt:
              - ActionsQueue
              - Arn
          batchSize: 1

resources:
  Resources:
    TriggerQueue:
      Type: "AWS::SQS::Queue"
      Properties:
        QueueName: "trigger_queue"
    ActionsQueue:
      Type: "AWS::SQS::Queue"
      Properties:
        QueueName: "actions_queue"

This configuration sets up two AWS Lambda functions that are triggered by messages from two SQS queues (trigger_queue and actions_queue). It also defines necessary permissions and environment variables.

Now to deploy the service on AWS, enter the following command in your terminal.

serverless deploy --region ap-south-1

Integrate with AWS SQS.

You can find the source code for the project here.

In this section, we will integrate our API with AWS SQS by creating a helper file for AWS configuration. This file will handle the setup for connecting to AWS SQS.

What is AWS SQS?

AWS Simple Queue Service (SQS) is a fully managed message queuing service that allows you to decouple and scale microservices, distributed systems, and serverless applications. It enables you to send, store, and receive messages between software components.

Steps to Integrate AWS SQS

Create AWS Helper File

We need to create a new file for AWS configuration. This helper file will contain the configuration for AWS SQS.

Create the file: In your project directory, create a new file named awsHelper.ts.
Add AWS SQS Configuration: Add the following code to awsHelper.ts to configure AWS SQS:

// Import AWS SDK
import AWS from 'aws-sdk';

// Configure AWS SQS
export const sqs = new AWS.SQS({
  accessKeyId: process.env.AWS_ACCESS_KEY_ID,
  secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
  region: process.env.AWS_REGION,
  apiVersion: process.env.AWS_API_VERSION,
});

You can find these credentials in the Configure AWS CLI section.

Use the AWS Helper in Your Handlers

In your handlers where you need to interact with AWS SQS, import and use the sqs instance from the AWS helper file.

// formController.ts

import { sqs } from './awsHelper';

let params = {
  MessageBody: JSON.stringify(workflowData),
  QueueUrl: process.env.AWS_SQS_URL as string,
};

// Send the message
sqs.sendMessage(params, (err, data) => {
  if (err) {
    console.log('Error', err);
  } else {
    console.log('Success', data.MessageId);
  }
});

After deploying and creating the AWS resources in the Deploy with Serverless Framework section,

Log in to your AWS Management Console.
Navigate to SQS: Search for “SQS” in the search bar and select the SQS service.
Find the TriggerQueue:

Look for the TriggerQueue in the list of queues.
Click on the TriggerQueue to view its details.
Find and copy the URL of the TriggerQueue.

Update Environment Configuration:

Update your environment configuration with the URL of the TriggerQueue you just copied as AWS_SQS_URL.
This URL will be used in your application to send messages to the queue.

Test the Integration:

Test by trying to send a message to the queue by hitting your API endpoint using Postman.

Deploy API Server on AWS EC2.

You can find the source code for the project here.

After thoroughly testing the API server and lambda functions, the next step is to deploy your API server on an AWS EC2 instance. Follow these steps to ensure a smooth deployment process.

Launch an EC2 Instance

Log in to the AWS Management Console.
Navigate to the EC2 dashboard and click on “Launch Instance.”
Give your instance a name and select the operating system (Ubuntu).
Choose an instance type, such as t2.micro (free tier) or t2.small.
Create a new key pair, which you’ll use to access the instance.
Under network settings, select to allow traffic from HTTPS and HTTP.
Review and launch the instance.

Expose the port 8080

On the EC2 dashboard, select your instance.
Click on the security group and edit the inbound rules and add a rule to allow traffic on port 8080.

Connect to Your EC2 Instance

Open your terminal and set the correct permissions for your SSH key:

chmod 700 Your_SSH_Key.pem

SSH into the machine using the EC2 instance URL:

ssh -i Your_SSH_Key.pem ubuntu@your-ec2-instance-url

Setup the project

Clone the Git repository:

git clone https://github.com/bhaskarcsawant/hubspot_workflows

Install Node.js:

// install nvm
curl -o- https://raw.githubusercontent.com/creationix/nvm/v0.32.0/install.sh | bash

// activate nvm
. ~/.nvm/nvm.sh

// install nodejs
nvm install 20

Install project dependencies:

cd your-project

npm install

Add environment variables:

// create a .env file
vim .env

// add following variables to the .env file and save and exit
MONGODB_URI=your_mongodb_uri
AWS_SQS_URL=your_trigger_sqs_queue_url
AWS_ACCESS_KEY_ID=your_aws_access_key_id
AWS_SECRET_ACCESS_KEY=your_aws_secret_access_key
AWS_REGION=your_aws_region
AWS_API_VERSION=latest

Start the server:

npm start

Visit your server in a browser at:

http://your_ec2_domain:8080 

// Eg. http://ec2-54-81-243-55.compute-1.amazonaws.com:8080/

After successfully setting up the server update the API URL in the lambda function’s environment variable and deploy the updated serverless service using the following command in your terminal.

serverless deploy --region ap-south-1

Test Your Workflow

Create a form submission using Postman.
Verify that a message containing your form submission is created inside the messages MongoDB collection.

Let’s see this in action. 😄

This is one possible architecture that allows you to build an automated workflow system. Using this architecture, you can create different types of triggers and actions, so that with a good frontend, your users can set up automated workflows without any coding.

Scalability.

Given the architecture is scalable, several improvements and additions can enhance its efficiency and performance. Let’s explore these enhancements and how to implement them effectively.

Connection Pooling

Problem: In the triggerHandler Lambda function, each invocation creates a new database connection, which can overwhelm the database at scale.
Solution:
Using EC2 Instances as Queue Workers: Instead of using Lambda functions, you can set up an EC2 instance as a queue worker. This allows you to use the same database connection across multiple requests, which can be scaled horizontally.
Using BullMQ: Another approach is using a service like BullMQ, which provides queue workers out of the box and processes queue messages asynchronously.

Autoscaling groups

Problem: As the load increases, manual scaling can be inefficient and may lead to downtime.
Solution: Implement autoscaling groups to automatically adjust the number of EC2 instances based on demand. This ensures that your application can handle varying loads efficiently.

Additional Points for Scalability

Load Balancer: Use an Elastic Load Balancer (ELB) to distribute incoming traffic across multiple EC2 instances. This ensures high availability and fault tolerance.
Caching: Implement caching mechanisms (like Redis or Memcached) to store frequently accessed data. This reduces the load on your database and improves response times.
Database Scaling: Use Amazon RDS with read replicas to distribute the read load across multiple instances.
Monitoring and Logging: Implement robust monitoring and logging using AWS CloudWatch or NewRelic. This helps you track the performance of your application and quickly identify and resolve issues.

If you find other ways to enhance scalability or encounter any issues during implementation, please feel free to comment or reach out to me. Your feedback will help everyone and is greatly appreciated. I’m always open to suggestions and eager to help you! Additionally, feel free to contribute to the codebase regarding any enhancements. Your contributions are valuable and welcomed! 😊

Follow Me for More Insights

Connect with Me

Twitter: https://twitter.com/Bhaskar_Sawant_
LinkedIn: https://www.linkedin.com/in/bhaskar-sawant/
GitHub: https://github.com/bhaskarcsawant

Stay tuned for more articles, and feel free to reach out if you have any questions or suggestions for future topics.

Cheers, thanks for reading! 😊

Happy coding! 🚀

This article was originally published on Medium. It is also available on Hashnode and Dev.to .

How to transfer data between tabs using a Chrome extension

Bhaskar Sawant — Fri, 14 Jun 2024 15:55:27 +0000

In today’s interconnected web, it’s common to need communication between different browser tabs, even if they come from different origins. Chrome extensions and the postMessage API make it easy to transfer data between tabs. You can create a Chrome extension using TypeScript to facilitate cross-origin communication. In this guide, I’ll show you how.

What You Will Learn

Setting up a basic Chrome extension with TypeScript.
Using postMessage API for cross-origin communication.

Setting up a basic Chrome extension with TypeScript.

First, let’s set up a basic Chrome extension project with TypeScript.

Follow these steps:

1. Create the Project Structure

I’ll be using this project structure throughout this project.

my-chrome-extension/
├── src/
│   ├── background.ts
│   ├── content.ts
├── manifest.json
├── tsconfig.json
└── package.json

2. Initialize thepackage.jsonand set up TypeScript.

Enter the following commands in your terminal.

npm init -y
npm install typescript --save-dev

Configure TypeScript by adding the following to tsconfig.json .

{
    "compilerOptions": {
      "target": "ES6",
      "module": "commonjs",
      "outDir": "./dist",
      "strict": true,
      "esModuleInterop": true
    },
    "include": ["src"]
}

3. Configuremanifest.json.

Create a file named manifest.json and add the following to configure the manifest.

This is the configuration file of your Chrome Extension.

{
    "manifest_version": 3,
    "name": "My Chrome Extension",
    "version": "1.0",
    "background": {
      "service_worker": "background.js",
      "type": "module"
    },
    "content_scripts": [
      {
        "matches": ["<all_urls>"],
        "js": ["content.js"],
        "run_at": "document_start"
      }
    ],
    "permissions": ["tabs"]
}

4. Create service_worker and content script.

Create src/background.ts in the folder structure, this will be the service worker of your Chrome Extension, it will keep running in the background.

Create src/content.ts in the folder structure, this will be the content script of your Chrome Extension, it gets loaded in every active tab in your browser. Add the following code to the content.ts .

document.addEventListener('DOMContentLoaded', () => {
    console.log('content.ts loaded');
});

5. Updatepackage.jsonScript to build your Extension.

  "scripts": {
    "build": "tsc && cp manifest.json dist/"
  },

Now enter npm run build in your terminal to build the Chrome Extension.

6. Load the Extension in Chrome.

Open Chrome and navigate to the Extensions page: Type chrome://extensions/ in the address bar and press Enter.
Enable Developer Mode: Toggle the switch on the top right to enable Developer Mode.
Load Unpacked Extension: Click on the “Load unpacked” button and select the dist folder of your project.

Your extension should be loaded, and the manifest, background script, and content script should be correctly set up.

To confirm open a tab, open the developer tools of your browser and click on the console tab, there you’ll see the console log which we wrote in the content.ts .

This way you can create a simple Chrome Extension.

Using postMessage API for cross-origin communication.

Now let us see how we can communicate between multiple tabs using postMessage API.

This tab inter-communication works not exactly but on a type of basic PubSub architecture in which background.ts which is our service worker works as a publisher and content.ts which our content script works as a subscriber.

When we send a message from one tab to the content script of that tab the content script forwards that message to the service worker and then the service worker forwards that message to the content script of all other active tabs.

Now, let’s code this up. 😄

1. Add Chrome types in the TypeScript project.

To use Chrome API in the TypeScript project we need to install the following package.

npm i @types/chrome

2. Set up the Content Script.

The content script will listen for the messages from the tab and send them to the background script then the background script will forward the messages to the content scripts of all other active tabs and then the content script will forward the message received from the background script to the tab.

 //content.ts : 

document.addEventListener("DOMContentLoaded", () => {console.log("Hello from content script")});

//this will listen to the message from the tab and send it to the background script
window.addEventListener("message", (event) => {
  if (event.source !== window || !event.data.type) return;
  const {type, data} = event.data;
  if (type === "FROM_TAB") {
    chrome.runtime.sendMessage({ type: "FROM_TAB", data });
  }
});

//this will listen to the message from the background script and send it to the tab
chrome.runtime.onMessage.addListener((message, sender, sendResponse) => {
  const { type, data } = message;
  if (message.type === "TO_TAB") {
    console.log("content script sending message to tab", message.data);
    window.postMessage({type:'FROM_CONTENT', data}, '*');
  }
});

3. Set up the background script (service worker).

//background.ts :
//this will listen to the message from the content script and send it to 
//the content scripts of all the active tabs
chrome.runtime.onMessage.addListener((message, sender, sendResponse) => {
  const {type, data} = message.data
    if (message.type === 'FROM_TAB') {
      chrome.tabs.query({}, (tabs) => {
        tabs.forEach(tab => {
          if (tab.id && tab.id !== sender.tab?.id) {
            chrome.tabs.sendMessage(tab.id, { type: 'TO_TAB', data  });
          }
        });
      });
    }
  });

Now save all the files and enter npm run build in your terminal to build the Chrome Extension.

Now go to the Extensions tab and refresh your extension.

To test this out will create a simple HTML page.

<!-- index.-->
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Document</title>
</head>
<body>
    <h2>Waiting for messages:</h2><br />
    <p id="texts"></p>
</body>
<script>
    let texts = document.getElementById('texts');
    window.addEventListener('message', (event) => {
        const {type, data} = event.data;
        if (type === "FROM_CONTENT") {
            console.log(event.data);
            texts.innerHTML += data + '<br />';
        }
    });
</script>
</html>

This will listen to all messages and print them on the screen. you can use plugins like Live Server to start this HTML file.

Let’s see this in action. 😄

By following this guide, you’ve learned how to set up a Chrome extension with TypeScript and enable seamless cross-origin communication between tabs using the postMessage API. This powerful setup allows for a richer, more interconnected user experience across different browser tabs. using postMessage API you can achieve things like scrapping, messaging and many more.

Follow Me for More Insights

Connect with Me

Twitter: https://twitter.com/Bhaskar_Sawant_
LinkedIn: https://www.linkedin.com/in/bhaskar-sawant/
GitHub: https://github.com/bhaskarcsawant

Stay tuned for more articles, and feel free to reach out if you have any questions or suggestions for future topics.

Cheers, thanks for reading! 😊

Happy coding! 🚀

This article was originally published on Medium. It is also available on Hashnode and Dev.to .