Sync product data from your e-commerce to your search or relevance engine with Nexus

Having a search engine that is not built in and pre-integrated in the e-commerce system is daily life for many e-commerce operators. Additionally, the search engine is often used for more than just search; it is also used for category listings and recommendations. In many ways the search engine becomes as critical a component as the e-commerce system itself.

A challenge everyone faces is syncing product data from the e-commerce system or the PIM to the search engine. In many cases you also want to sync data from other systems into the search engine. You want to send in stock information and price, you want to send in pure CMS pages to make them searchable, and you want to send in product images and product data.

This is a harder challenge than it might first appear, especially in cases where the search engine is used for more than just search. When you send price and stock information to the search engine, it becomes very important that updates of these are sent to the search engine within seconds after they happen. It is easy to take for granted that it should just work, but many integrations still build on nightly runs or scheduled runs a couple of times a day.

Common pitfalls

A common pitfall for solving this is hoping that a finished plugin between the e-commerce system and the search engine will suffice and work. The problem with these plugins, which are often built by either the vendor of the e-commerce system or the search engine, is that it is not their core business. It is something they have to build to be able to say in sales meetings that there is an integration. They often work OK to get started with, but as soon as you have wishes that do not fit into a sales demo, you hit limitations. The place between the e-commerce system and the search engine is a bit of a no man's land that neither of the vendors really wants to be in, so the result is usually accordingly.

Another common pitfall is thinking you can use serverless technology with cloud events like Azure Functions or AWS Lambda. In theory it sounds like a perfect approach. The e-commerce system generates an event saying a product is updated, you listen for such an event in your serverless function and update the search engine. The problem is that often sweeping updates happen on a large amount of products, generating a large amount of events at once. Listening for and processing these events one by one will take significantly longer than if you process them in batch. It is also not uncommon for several events to happen more or less at the same time on the same product. If these are processed one by one, the integration takes much longer than necessary.

In other cases synchronisation of different events is needed before it can be sent to the search engine. You do not want to send the product to the search engine until it has been given a price, for example. If you process these events one by one in a serverless architecture, it becomes a growing complexity in determining what extra things need to happen to handle this kind of synchronisation.

Nexus

Nexus is a free-forever tool that Commerce Mind has developed with focus on simplifying and ensuring the integrations and data flows that are daily life for an e-commerce operator. We have boiled down our experience and knowledge into a tool that we offer so you can sleep a little better and stop worrying about whether your data is flowing correctly between systems.

What we start by doing in Nexus is creating a queue. The first question we are faced with is what data we want to store in the queue message. Generally there are two ways. The first is that the message contains an id for, for example, the product, which we then use to fetch the product data when the message is processed. The second way is to let the message contain the product data we want to send to the search engine. Both these ways have their pros and cons, but you get more out of Nexus if you store the data in the message. For example, Nexus can skip updates of the product that do not change the data that is sent into the search engine. Another advantage is that it is faster to process the message because we do not first have to fetch product data.

Our to-begin-with empty queue message thus looks like this:

[QueueMessage("product", IdempotentMessages = true)]
public class ProductChangedQueueMessage : IQueueMessageWithId
{
    public string? Id { get; set; }
}

Because each message has an id and we know we have a limited number of products, we turn on Nexus saving processed messages. Even if we have millions of products, it is still a limited number that a Nexus queue can handle without problem.

The interesting thing about saving messages instead of throwing them away when they are processed is that it lets us do things like compare to the previous message to make smarter decisions. A common thing in the e-commerce world is that the URL for a product is decided by the product's name, and if the product's name changes, redirects need to be created from the old to the new name, for example. Or handling legal requirements around price history by saving the prices in a Nexus queue and automatically knowing what the previous price was.

So let us assume we store data like product name, description, images and price in the message itself. The next question we are faced with is which id we should use for the queue message. One candidate is the internal product id from the e-commerce system. But it is a fairly poor candidate because only the e-commerce system knows it. If in the next step we want to listen for changes in the ERP that does not have access to this id, we will not be able to merge those updates. Instead we need to use an identifier that is shared between systems, typically product number or item number.

Our queue message now looks roughly like this:

[QueueMessage("product")]
public class ProductChangedQueueMessage : IQueueMessageWithId
{
    public string? Id { get; set; }
    public ProductInformation? ProductInformation { get; set; }
    public ProductPrice? Price { get; set; }
}

public class ProductInformation
{
    public required string? Name { get; set; }
    public required string? Description { get; set; }
}

public class ProductPrice
{
    public required decimal Price { get; set; }
    public required string Currency { get; set; }
}

The interesting thing about splitting the product information as name and description into its own part of the message and the price being its own part is that they can then be sent into the queue independently of each other by taking advantage of a queue being able to receive incomplete messages. That is, we can send in product information and price from different places, and Nexus guarantees that the data from the different sources is merged correctly in the same message.

The next step is to start filling our queues with data. Nexus will automatically make an IEnqueuer<TMessage> available that we can use to add messages:

class ProductService(IEnqueuer<ProductChangedQueueMessage> enqueuer)
{
    public async Task EnqueueProductChangedAsync(string productNumber, string name, string description)
    {
        await enqueuer.EnqueueAsync(
            new ProductChangedQueueMessage 
            {
                Id = productNumber,
                ProductInformation = new ProductInformation
                {
                    Name = name,
                    Description = description,
                }
            }, 
            new EnqueueContext { PartialMessage = true }
        );
    }
    
    public async Task EnqueuePriceChangedAsync(string productNumber, decimal price, string currency)
    {
        await enqueuer.EnqueueAsync(
            new ProductChangedQueueMessage 
            {
                Id = productNumber,
                Price = new ProductPrice
                {
                    Price = price,
                    Currency = currency,
                }
            }, 
            new EnqueueContext { PartialMessage = true }
        );
    }
}

Now that we have data in the queue, it is time to create a job that can process the queue:

public class ProductQueueJob(IPreviousMessageProvider previousMessageProvider) : IScheduledQueueJob<ProductChangedQueueMessage>
{
    public string DefaultSchedule => CronSchedule.TimesPerMinute(10);

    public async Task<ProcessResults> ProcessMessageAsync(ProductChangedQueueMessage message, CancellationToken cancellationToken)
    {
        var previousMessage = previousMessageProvider.GetPreviousMessage(message);
        if (previousMessage?.ProductInformation?.Name != null &&
            message.ProductInformation?.Name != null &&
            previousMessage.ProductInformation.Name != message.ProductInformation.Name)
        {
            await CreateRedirectAsync(from: previousMessage?.ProductInformation?.Name, to: message.ProductInformation?.Name);
        }
        
        if (message.ProductInformation == null || message.Price == null)
        {
            // Once the missing data comes in the processing job will run again so we mark the
            // message as processed
            return ProcessResults.Processed("Skipping processing of incomplete message");
        }
        
        await SendProductToSearchEngineAsync(message);

        return ProcessResults.Processed();
    }
}

Here we see how we can easily read up the previous message and compare what has changed to act on specific changes. The previous message is always the latest message that was processed by the Nexus job. That is, you do not risk missing a change if the job was paused and several changes came in during that time.

In this case we process each message one by one, which is the simplest way to do it, but for large amounts of products it will take much longer than processing several messages at a time. With Nexus it is of course easy to process messages in batch instead:

public class ProductQueueJob : IScheduledBatchQueueJob<ProductChangedQueueMessage>
{
    public string DefaultSchedule => CronSchedule.TimesPerMinute(10);
    public int BatchSize => 100;

    public async Task ProcessMessagesAsync(QueueMessageBatch<ProductChangedQueueMessage> batch, CancellationToken cancellationToken)
    {
        await SendProductsToSearchEngineAsync(batch.Messages);
    }
}

Now we listen for changes to product information and prices and add them to the Nexus queue, and we have a job that processes the queue 10 times a minute. The reason queue jobs in Nexus are scheduled rather than running as soon as a queue message comes in is to be able to wait for a larger amount of messages and take advantage of the fact that it is faster to process many at a time. But nothing prevents you from scheduling the job to run every second.

What we have left is a way to run a full sync. To take all product data and all prices and run again.

public class EnqueueAllProductsJob(IEnqueuer<ProductChangedQueueMessage> enqueuer, 
                                   ErpService erpService, 
                                   ProductDataService productDataService) : IScheduledJob
{
    public string DefaultSchedule => CronSchedule.NotScheduled();
    
    public async Task<JobResults> ExecuteAsync(CancellationToken cancellationToken)
    {
        await enqueuer.EnqueueAsync(
            (await erpService.GetAllPricesAsync())
                .Select(price => new ProductChangedQueueMessage 
                {
                    Id = price.ProductNumber,
                    Price = new ProductPrice 
                    { 
                        Price = price.Price, 
                        Currency = price.Currency,
                    }
                }),
                new EnqueueContext { PartialMessage = true }
        );
        
        await enqueuer.EnqueueAsync(
            (await productDataService.GetAllProductsAsync())
                .Select(product => new ProductChangedQueueMessage 
                {
                    Id = product.ProductNumber,
                    ProductInformation = new ProductInformation 
                    { 
                        Name = product.Name, 
                        Description = product.Description,
                    }
                }),
                new EnqueueContext { PartialMessage = true }
        );
    }
}

Here too we take advantage of Nexus being able to receive incomplete messages, because the ERP can have prices for products we do not have product information for yet, and vice versa. Even though we cannot process these messages until the other data comes in, we still want to get the data into the queue.

How long this job takes depends entirely on how quickly it can read data from the other systems. Nexus can without problem receive tens of thousands of messages per second.

The effect of this job is that only the products whose data has changed since Nexus processed them last will be synced, because we keep track of the previous version of the message. But if you want to force a complete sync anyway, you can just update all queue messages' status via the admin interface that is included in Nexus:

Why is my product not showing on the site?

Before the product data is sent to the search and relevance engine, we want to validate that the data is correct. We have already added validation ensuring that the product has a price before it is sent, but often more logic than that is needed. We want to make sure the product has one or more images, one or more categories and so on. This logic can easily be added in the job.

A common question from the business at many e-commerce companies is "Why is product X not showing on the site?" and often it is because it has gotten stuck in some validation. Let us make it easy for the business to see this for themselves:

public class ProductQueueJob : IScheduledQueueJob<ProductChangedQueueMessage>
{
    public string DefaultSchedule => CronSchedule.TimesPerMinute(10);

    public async Task<ProcessResults> ProcessMessageAsync(ProductChangedQueueMessage message, CancellationToken cancellationToken)
    {
        if (string.IsNullOrEmpty(message.ProductInformation?.Description))
        {
            return ProcessResults.CustomStatus("Invalid", "Missing description");
        }
        
        await SendProductToSearchEngineAsync(message);

        return ProcessResults.Processed();
    }
}

In this case we check whether there is a description, and if it is missing the message ends up in the status "Invalid" with the description "Missing description". It then becomes easy for anyone to filter forth all messages in the status Invalid in the admin interface to see what has gone wrong:

Summary

Even though it is a good use case for Nexus to sync product data to a search engine, it is hardly the only one. In this article we have gone through a number of the features that make Nexus special, and perhaps above all how easy Nexus makes it to avoid the common pitfalls. Want to automatically re-sync all products in a category when the category name changes? For most people this requires a manual full sync. With Nexus it becomes easy to automate.

There is much more functionality to explore beyond this article, such as letting jobs have parameters, using virtual queues, taking advantage of Nexus Functions when something simpler than a queue is needed or seeing all previous versions of a queue message.

Does this sound like it could help you with your data flows? Get in touch. Nexus is free forever and you get full access to the code via an open source licence.

Follow us on LinkedIn for a notification of when we release the article on how Nexus handles re-syncing the product when future, date-driven prices are activated.