email-generation.svg

Graph API

Product scaling

## Problem
Each user from the marketplace of 10 million users has an individual set of products that are recommended for purchase. Actual up-to-date data on products recommended for purchase must be collected, stored, and delivered to each user in a “Resources you may like” (RYML) email. A way to generate RYML emails with the relevant content and send them to users on a regular basis has to be developed.

### Goals and challenges
- Gather actual up-to-date lists of products recommended for purchase for each user.
- This data must be stored and then delivered to all subscribed users via “Resources you may like” (RYML) emails on a regular basis.
- Delegate email creating and sending sequence functions to email sending service.
- Maintain business logic of collecting recommended data and providing it to the email sending service.

## How Wise Engineering helped
The Recommendations Email Generation Pipeline solution was developed. It allows us to gather accurate information about products recommended for purchase for each user and store this info afterward for a needed period of time.
Results
- Information about recommended products is being accurately collected.
- Email Pipeline solutions gives an ability to transfer collected information regarding 10 million users to the email sending service within only 5 hours.
- Email sending service gets the information if a particular user has any recommended products, and can easily send that list in email messages on a regular basis.

## Background
There is a marketplace with 10 million users and each user has specific products that are recommended for their purchase. This information must be collected and delivered to the user on a regular basis.'

## Implementation process

To begin with, we focused on how to orchestrate composition of microservices and to queue extracted products’ recommendations data. Next, we’ve evaluated options for transforming this data into a format that is readable for third-party services. Subsequently, we’ve designed a way to load this data into an email sending service.

The following tools and techniques were applied to achieve these goals:

1. PHP script to go over all users to group them and chunk by task type (if it has recommendation or not). After that, assigned user ID chunks into queue.
2. PHP workers process the queue with user’s ID data and retrieves recommendations for each of them.
3. Elixir Graph API application exposes data on recommended products to PHP workers. Next, it processes products’ recommendation data and applies some internal rules and limitations on it.
4. Python REST API application generates product recommendation data for a particular user ID on demand.

## Solution

We came up with the solution and called it “Email Pipeline.” It starts with running PHP script to go over all users and creates a bunch of async tasks. Each task contains information about:
- User IDs that should be processed
- What kind of operation needs to be done

It may happen that some users do not have any recommendations. In this case, we have to clean up their data on the third-party email service. The rest of them should be updated with new recommendations data. This diversification happens on the fly and group users based on the type of task.

The tasks creation process also has a time ramp up feature. Task processing jobs are heavy and time costly operations. To mitigate that, pipeline services run on Kubernetes’ cluster with autoscaling, depending on their load. So, to make sure that autoscaler has enough time to ramp up the cluster size, we create tasks with ramp up time. We increase the rate of tasks being created in 1 hour. After that time, we spawn tasks as fast as we can as all services are scaled and ready to process increased loads created by task processing jobs.

While tasks are being created, workers start to pick them up and process. The process starts with defining the type of task. In case a user doesn’t have any recommendations, it goes straight to a third-party email service and nulls recommendation data for this user. If the user has products recommendations data, then we need to fetch it fir

To fetch this data, we call Elixir Graph API service which represents a middleware between tasks, or in our case, front-end and recommendation storage. Recommended products data itself is stored in a non-precalculated state hold by Python microservice. This service can multiply matrices on the fly and calculate scores for possible product recommendations and return the data to the caller. After Graph API calls and receives data from Python microservice, the data needs to be processed and cleaned up as this service just keeps the list of recommended products and doesn’t know anything about the user. The cleanup process means removing items that were bought by a user already, were wishlisted, added to the cart, or even seen somewhere else on the site. This information is accessible by Graph API so we can easily clean up recommendations we don’t want to be served to the user next time.

We now have information about the set of products recommended to each user in the task, regardless whether it is nulled data or new recommendations. The only step left is to transfer this data to a third-party email sending service. We will use API of the third-party email sending service to perform this.

As a result, the email sending service has the information if a particular user has any recommended products, list of recommended products if there are any, and can easily send that list in email messages on a regular basis.

## Results

The Recommended Products Emails Generation Pipeline gives the ability to transfer collected information about relevant recommended products concerning 10 million users to the email sending service in 5 hours. This allows us to send “Resources You May Like” emails on a daily basis. All tasks that are related to email composing have been delegated to a third party email sending service. This gives us an opportunity to send emails to all the users with one click after the info on recommended products has been imported.


Learn how our team developed recommendations email generation pipeline and what results we achieved.

Recommendations email generation pipeline

scheduled-emails.svg

## Why?

Our main target is to develop a system for sending large numbers of emails and then tracking users’ interaction with a received mail (e.g. opens, link clicks, unsubscribes, etc.) Users with large amounts of daily sends prefer use scheduling for sending use.

## Why we need scheduling

Using scheduling, instead of send-immediately flows, gives us several advantages.
- Better control for sending process
- If a user needs to send emails after his workday is finished, then he won’t be able to monitor the sending process and fix them if something goes wrong. With scheduling, users can prepare all emails for sending before the actual send date, and our ESP (Email Service Provider) handles the rest.Less resources required
- If you already have content for emails that you need to send (for example tomorrow), you are able to send the emails gradually. For example, if you need to send 5M emails at 8 PM you can start sending at 10 AM and continue sending in small batches the whole day. We collect the emails from our side and cover the delivery in time.No personal CRON.

If you have a notification email that needs to be sent after some time (for example, a subscription reminder), just send the notification to our ESP and we cover the scheduling. You don’t need to handle that case with personal Cron scripts.

## Problems that we faced when building scheduling

We propose pretty simple API for working with scheduling, but it’s not easy to organize simultaneously sending even 1M emails. We needed to think about scalability and our system in the future. We had to solve several problems.

- Proper storage for scheduled emails
- Limited throughput of the network for one process
- Split scheduled emails that were ready to send for different workers
- Syncing statuses of scheduled emails and avoiding duplication on send

## Proper storage for scheduled emails

We should able to filter “ready for send” emails as fast as possible among large amounts of records that don’t need to send yet. At the time we already used ClickHouse database as storage for our analytics events (unsubscribe, clicks, open, etc.).

Clickhouse is a column-oriented database that is used mostly for analytics. But with the read/write throughput that ClickHouse provides for us we are able to use the solution as scheduling storage.

The most valuable storage requirements for us were the following:
- Local and distributed joins
- Data compression
- Parallel and distributed query execution
- Cross-datacenter replication
- High availability
- SQL support
- Fetch speed not less than 500K email records per second.

## Performance testing

We made some performance tests and got results that completely satisfied our requirements:

- Read speed is 1.2 GB/s
- Write speed is 50 to 200 MB/s
 - Finding 335K scheduled emails on a dataset with the size of 5M takes 0.061 second

As a result, with network latency for transfers, we were able to fetch 3.5M emails per second. That was a very good result for us.

## Limited throughput of the network for one process

We resolved the problem of choosing our data storage and then faced another issue. We used NodeJS as our main platform, the platform, together with SQS queues, didn’t provide the desired performance for us. Then we wrote simple prototypes for testing using Golang and Elixir (we hoped that Erlang with its lightweight threads helps us) we got some throughput increases, but still weren’t reaching the desired results and scalable solutions.

Results of our benchmark’s queue inserting time for 50K messages to queue was as follows:
- NodeJS 1 thread = 95.408 sec
- NodeJS 2 thread = 75.757 sec
- Go = 35.564 sec

Elixir was about 2 times slower than NodeJS with 2 threads and the same kind of problem.
So obviously, we needed to find ways to scale horizontally.


## Split scheduled emails that are ready to send for different workers

Obviously, we needed to make our scheduling scalable and found a way to split all scheduled emails for workers. Workers should have the same role and know nothing about each other. It’s allowing us to run more workers on more servers when we need to process the largest amount of emails.

When workers are fully independent from each other, we decided to create one more component in our system.

The component is named Ranger and is responsible for splitting ranges for our workers. Each range shouldn’t intersect with other ranges.

With Ranger and Clickhouse working together, we created the following architecture for our scheduling feature:

1. Emails send to REST API, and if a parameter with a future send date exists, API puts the email to Messaging Queue.
2. Write to the consumer processing the Queue and save all emails inside Clickhouse database.
3. Ranger runs query for fetching emails and the query contains the following steps: Records filtered by send date and each time we fetch all records that are ready in that moment. Filter records by time-created range, we need only those that weren’t affected by the previous step. For that, we fetch only records that have a date greater than biggest date from previous execution. Aggregate results by CreatedTime and COUNT function.
4. Set first fetched date as DateFrom for query.
5. Wait for full batch size and set last date as DateTo.
6. Update Date to as last processed range.
7. Send range to Queue for processing by worker.
8. Consumer pulls a Queue with ranges and fetches by CreatedTime emails for sending to user.
9. Each email sends to target user.


![scheduled emails scheme](https://wise-site.fra1.digitaloceanspaces.com/scheduled_emails_scheme_94e7ec9efe.svg)

## Avoid duplication

AWS SQS service promises not less than one delivery for a Standard Queue. this means that sometimes we can obtain one message on two workers and send our email twice. Since each message can contain multiple emails (our current batch size contains 5K) we need to organize duplicate handling.

For that, we have the Redis database and we store each SQS messageID as a Redis key. The keys have an expiration date that's equal to message visibility timeout in the Queue. When a consumer obtains a new message from SQS, he checks for an existing messageID in Redis. If the message already exists, the consumer skips this one.

## Conclusion

Using ClickHouse and AWS Simple Queue Service, we build a scalable scheduling system for delivering emails. These flows and tools can be applied to any actions that could be scheduled, not just for emails.

At the moment, we can fetch 335K records per second, filtering by ready date takes 0.1 seconds for 5M database set (including network latency), and workers that process the emails can easily scale horizontally for the desired throughput.


Check out how our team built a scalable scheduling system to deliver emails via ClickHouse and AWS Simple Queue Service.

How to send large amount of scheduled emails simultaneously

s3-file-upload.svg

## Problem

Users could not upload all their products to the marketplace because most of them were files that are larger than 200 MB or were video files. Both were not supported for upload; there was a size limitation for digital files, and video files were not supported at all.

## Goals and challenges

- Allow users to upload files that are larger than 200 MB.
- Enable users to continue file upload after the network error from user side has occurred.
- Scan uploaded files for malware.

## How WE helped

Direct upload of files to Amazon S3 was designed and successfully applied in the existing service (marketplace) within a shorter period of time than the upload system that was used before: half a month compared to 2 months. Moreover, it also enhanced security for downloaded files.

## Results

- Users could upload files up to 500 MB or up to 1 GB (depending on association with a specific user group).
- New items (Video Files) became available for upload.
- S3 server usage decreased the time and resources required for server maintenance compared to the hardware servers that were used before.
- The project received enhanced security of downloaded files.

## Background

Existing service allowed users to upload files and attach them to products. However, the service put a solid limit on the size of files that were selected for upload. As a result, it was not possible to upload a file that is larger than 200 MB. Also, it worked in a way that when a file upload failed (due to the network error), the user was forced to start the entire process all over again. A major problem was that many users had files usually larger than 200 MB. Consequently, there was a lot of requests to upload such types of files. Besides, the NGINX module which was used for file upload was very old and unsupportable.

## Implementation process

### Old upload schema

Previously, when the user uploaded a file, we used an old NGINX module. This NGINX module simply read all the content of the file and stored it on our server in a temporary directory with a temporary file name. The following problems emerged:

1. This mechanism read the whole file within one operation. This approach didn't work correctly for the big files and caused a significant load on our servers.
2. The unique temporary file name was not, in fact, unique. Sometimes, when we had lot of users on the website, we had conflicts when two or more users uploaded files with the exact same temporary file name.
3. Our upload progress bar displayed the upload process only. We could not add a progress bar for virus check, consequently users just saw a stalled progress bar while we checked the file for viruses. Also, we generated preview files, which were a reduced version of the original user’s files; because of this, users could not monitor the progression of the upload process.
4. After the upload, we stored all files on our own data servers. It required more attention from our side and resources to maintain them. Moreover, it made our codebase more complicated since we required some extra fields in the database in table item properties; we called it mask of servers, which contained information where a file for a specified item was stored. It increased complexity to select all items which stored files on specific data servers only.
5. This upload process was not secure:
- If someone hacked our system, they could catch and replace files.
- We checked files for viruses by Cron after the upload was completed. During the time between upload and check, an infected file could corrupt our system and harm other users who later bought the product and downloaded the infected file.

![direct s3 file upload old scheme](https://wise-site.fra1.digitaloceanspaces.com/direct_s3_file_upload_old_scheme_a8fadb287c.svg)

## Solution

### New upload schema

We decided to upload files directly to S3 and no longer store them on our servers. Also, we’ve chosen to use a 3rd party JavaScript library to upload files to S3 directly instead of creating our own library. We selected [https://github.com/TTLabs/EvaporateJS.](https://github.com/TTLabs/EvaporateJS. ) This library had already implemented and tested features for all known web browsers:

- Secured direct S3 file upload.
- Resume failed file upload.

Now all files were uploading to S3 into a temporary bucket for each user’s item. Also, we used unique paths and file names. We decided to split files into a smaller parts before upload and run uploads to S3 in parallel threads, each part was signed by our server so it couldn't be caught and replaced by someone else. Now when a user selects a file to upload to S3:

1. Evaporate JS uploads a file into a temporary bucket directly. During the upload process, it makes a request to our servers only for signing each part of the file. This is done to avoid cases when somebody can catch part of the uploaded file and replace it.
2. After uploading files into the temporary bucket, we start to process the files in the following order:
   - Check for viruses: If a virus is detected, we stop processing the file, move it into a quarantine bucket (it clears automatically after 4 days), and notify the user about this event.
   - Generate preview files.
   - Collect information about the file (size, checksum, type of file, etc.).
3. User fills in required text fields and clicks on submit button.
4. Process push job of funneling item into a queue:
When the user wants to create several items, we organize all the processes into a queue. Once the first item is submitted for creation/update, the user is able to select the next one and doesn't have to wait until the create/update process is finished for the first item. Once the item is created/updated, we display a message to inform the user.
5. In the background, worker process takes jobs from the queue and performs the next job in parallel threads: Save item details into database.
6. In case the background saving process has failed (network issue, Amazon issue, etc.), the queue process always restarts failed tasks a couple of times.

![direct s3 file upload old scheme](https://wise-site.fra1.digitaloceanspaces.com/direct_s3_file_upload_new_scheme_b28e226c13.svg)

## Results

The new upload process was designed and successfully applied. We gave our users an opportunity to upload files up to 500 MB. For premium users, we’ve added a new feature – files upload up to 1 GB. Also, we decided to allow upload of video files as a new item. This upload process was successfully applied in the existing service (marketplace) within a shorter period of time than the upload system that was used before: half a month compared to 2 months. The S3 server also was easier to maintain, and took less time and effort to maintain, compared to the hardware servers that were used before. Moreover, it also reinforced security for downloaded files.

Wise Engineering team helps businesses solve complex tech challenges like [migration to cloud](https://wise-engineering.com/expertise/cloud-migration/), [internal search integration](https://wise-engineering.com/expertise/elasticsearch-integration/), optimization for high load, and more. Check out our expertise in [custom software development](https://wise-engineering.com/expertise/) and contact us to discuss your future project.



Learn how our tech team developed a direct upload system to S3 and allowed users to upload large files to the marketplace seamlessly. 

Direct S3 File Upload | Wise Engineering Blog

Learn how we developed a direct upload system to S3 and allowed users to upload large files to the marketplace seamlessly. 

Direct S3 file upload

notification-system.svg

## Problem

The possibility to send a high number of emails without keeping your own server for email delivery.
Goals and challenges
- Track user interaction with the email (up to 10K notifications per minute).
- Reduce load on 3rd party app that sends high number of emails daily.
- Webhook receiver’s downtime shouldn’t affect main application flow.
- System has to be scalable and guarantee to deliver within one day, when hook target isn’t available.
- Convenient and timely email delivery status reporting and displaying this info on a dashboard.
- Service is easy to use and customizable to meet changing requirements.
- Availability and comprehensiveness of documentation.

## How WE helped

We've implemented the hooks delivery system with Amazon Simple Notification Service (SNS) and proxy server that processes, subscribes, and redirects requests. Little coding was required and responsibility of webhook delivery is assigned to an external service.

### Results

- Email delivery system with the user’s activity tracking feature that processes a high number of events (up to 20K notifications per minute).
- Information about user interaction with the email is captured and stored.
- Redelivery solution is available when webhook receiver is unreachable.
- System is separated and does not affect main app services.
- Solution is easy to scale.
- 2 times higher level of throughput.

## Implementation process

To achieve project objectives, we’ve considered several implementation options:
- Naive implementation.
- Implementation via SQS queue.
- Webhooks delivery SNS notification.

We have the following components that are responsible for events in our system:
1. Event consumer.
2. Pull queue and decide what we need to do with events.
3. Event storage.
4. Store all events in our system.
5. 3rd party app.
6. Have to be informed about events related to the app account.

## Native implementation

The simple way to inform a 3rd party API is just to make HTTP request the hook URL directly from the consumer, as displayed on the following diagram:

Pros:
- Easy to implement.

Cons:
- Main application flow is affected, thus if a 3rd party app won't respond, we will receive an error in our internal consumer.
- Consumer cannot be scaled separately from webhooks which are sent to 3rd party app.
- Impossible to implement webhooks redelivery correctly.

Accordingly, we came to the conclusion that this approach will not give us expected results or satisfy our needs in a comprehensive manner.

Therefore, we’ve started to consider other options.

![Native implementation](https://wise-site.fra1.digitaloceanspaces.com/webhoooks_native_implementation_2f8d52c370.svg)

## Implementation via SQS Queue

We use a lot of AWS services at the moment and already have a bunch of SQS Messaging Queues. The main idea for the solution it is to use a separate queue for hooks (webhooks) and consumer to process this queue. Take a look at the schema that shows how SQS queue works:
In this manner our main app flow does not depend on the working state of an external app and we are able to implement retries via SQS queue.

The flow of a retry is as follows:

1. Pull webhooks events from queue.
2. Send HTTP request with events in the body to 3rd party app.
3. If request is completed successfully, then remove message from queue.
4. If request has failed, retry with the specified delay and send a message once again.
5. Set maximum receive count that equals delivery retry count.

![Implementation via SQS Queue](https://wise-site.fra1.digitaloceanspaces.com/webhooks_implememntation_via_SQS_queue_abcecd5e9e.svg)

The flow is displayed on diagram below:
Pros:
- Main application flow is not affected.
- Redelivery can be easily implemented.
- Independent main event consumer can be scaled.

Cons:
- More complex to implement.
- Max redelivery delay is 15 minutes (since max delay is 15 minutes in SQS queues), thus we can’t make a logarithmic retry for our messages.
- Logic of retries and responsibility for webhook delivery has to be managed manually.

This option was way better, however, not all of our primary goals would be met by implementation via SQS queue.

![Implementation via SQS Queue general scheme](https://wise-site.fra1.digitaloceanspaces.com/webhooks_implememntation_via_SQS_queue_general_c43aaf6c6e.svg)

## Webhooks delivery by SNS Notification

Another way to implement webhooks delivery is by using AWS SNS. The service was developed for delivering notifications and can send them via HTTP, the same way as webhooks do. Also, if notification delivery has failed, SNS is responsible for retry delivery, and we can use linear or logarithmic function to manage retry frequency.

This solution will give us desirable outcomes; however, it cannot be applied due to these two reasons:

1. SNS send confirmation requests to subscribe endpoint (webhooks destination).
2. SNS provides redundant metadata to delivered messages.

The SNS was designed for communication between different components within one system and not for communication among different components of different systems with different owners. Hence, we need a proxy component inside our system that SNS can communicate with

Let's take a look at the scheme of proposed webhooks implementation using SNS:

![Webhooks delivery by SNS Notification](https://wise-site.fra1.digitaloceanspaces.com/webhooks_delivery_by_SNS_notification_ed2086657b.svg)

Flow of hooks delivery using SNS:

1. The consumer sends events to a certain SNS topic.
2. We subscribe a 3rd party app to the topic and the URL should contain an endpoint that we need a redirect for.
3. When the proxy receives a request for subscription confirmation, it will send a request to SNS with a proper confirmation token.
4. When the proxy receives a request with a notification (an event, in our case), it will format event data and redirect to a URL that contains the URL path part.
5. All responses from the hooks’ endpoints pass as responses to SNS requests.

Schema of proxy API is shown below:

Pros:

 - Main application flow isn’t affected by the state of the hooks’ receiver application.
- The responsibility for message delivery is shifted to SNS.
- Redelivery strategy is already implemented by AWS.
- The simple proxy has good throughput, very lightweight and can be scaled.
- Little code is needed (less code, fewer bugs).

Cons:
- The proxy has to be developed.

![Webhooks delivery by SNS Notification scheme 2](https://wise-site.fra1.digitaloceanspaces.com/webhooks_delivery_by_SNS_notification_scheme2_1eb5d905fe.svg)

## Solution

After considering each implementation option in terms of benefit it will bring, and required resources for its execution, we’ve chosen to go with the webhooks delivery by SNS notification. With SNS and the simple proxy server, we’ve implemented the hooks delivery system with minimum coding involved and shifted responsibility of webhook delivery to an external service. As a result, we are able to send about 20,000 events per minute without any scaling, which is even higher compared to the planned 10,000 events per minute. The major bottleneck is the proxy, but we designed this one for scaling so throughput can be easily increased.

## Additional info

We wrote our proxy using Node.js platform. The proxy hosted on EC2 T2 small instance and can process about 4,000 concurrent requests with throughput of about 400 req/sec. The proxy doesn’t have any interaction with the database since all data for redirects are passed in URLs. This gives us an option to do a horizontal scaling. If we place our proxy behind a load balancer, then we will be able to add more server instances and as a result obtain two times higher throughput.

Vertical scaling won’t give us the same effect. When we scale node processes within one server instance, we can get about 30% higher throughput for each additional processor kernel. Therefore, horizontal scaling is a more suitable and effective approach.

We don’t need to worry about scaling SNS because AWS does it automatically, we only have to send the proper count of messages.


Here’s how we implemented the hooks delivery system with Amazon Simple Notification Service (SNS) and a proxy server.

Wise Engineering Blog

Recommendations email generation pipeline

How to send large amount of scheduled emails simultaneously

Direct S3 file upload

Creating webhooks notification system