Integrate with an external search engine

1. Identify Products per Store via Product Selections:

• Query all your Stores.
• For each Store, retrieve its associated active Product Selections to determine the exact active product assortment.

GET /{{project-key}}/stores?sort=id asc&limit=50&withTotal=false&where=id > "00000000-0000-0000-0000-000000000000"

2. Gather unique Product IDs:

• Create a deduplicated list of all Product Selection IDs while still mapping them to the Stores.
• For each Product Selection, paginate through its assigned Products to get a list of all relevant Product IDs. To do this, you will need to query the Product Selection by ID and use a predicate cursor to get results greater than the Product ID in the last response.

GET /{{project-key}}/product-selections/00000000-0000-0000-0000-000000000000/products?limit=1&where=product(id+>+"00000000-0000-0000-0000-000000000000")

3. Fetch base Product data:

• Using the list of Product IDs, fetch the corresponding published Product Projections in batches. Use a where clause like id in ("id-1", "id-2", ...) and sort by id for efficient pagination.
• Deduplicate the request for Product ID’s, so that you don’t fetch the same Product multiple times.

GET /{{project-key}}/product-projections?where=id in ("id-1", "id-2")&staged=false&sort=id asc&withTotal=false

Example Request:

GET /{{project-key}}/product-tailoring?where=product(id in ("f77df92a-d424-4199-9b7c-e2b93a693ba2", "155bbbcb-1f93-4e66-a708-632d7ce8d5c7"))

Drawback: This method is only efficient if most of the Products in your list are actually tailored for the target Store. If your Project has a large catalog where only a small percentage of Products are tailored (for example, 10%), this query will result in many lookups that return no data, leading to wasted API calls and slower performance.

The key is to filter by store (optional) and published, then use a sort parameter to paginate through the results.

Here are two recommended pagination strategies.

Initial Query (First Page): The predicate filters for published Product tailorings in a specific Store and sorts by ID.

GET /{{project-key}}/product-tailoring?where=store(key="your-store-key") and published = true and &sort=id asc&limit=30

OR

Store key is optional.

GET /{{project-key}}/product-tailoring?where= published = true &sort=id asc&limit=30

Subsequent Queries (Next Pages): Use the ID of the last item from the previous response to fetch the next page (Store key is optional).

GET /{{project-key}}/product-tailoring?where=store(key="your-store-key") and published = true and id > "{lastIdFromLastResponse}"&sort=id asc&limit=30

Initial Query (First Page):

GET /{{project-key}}/product-tailoring?where=published=true&sort=productId asc&sort=id asc&limit=30

GET /{{project-key}}/product-tailoring?where=published=true and ((productId = "a42d12f2-..." and id > "e9ae13b3-...") or (productId > "a42d12f2-..."))&sort=productId asc&sort=id asc&limit=30

4. Enrich with additional data (for example, Categories)

• The Product Projection contains Category IDs, but not their names. You will need to fetch Category data separately and enrich your Product data with the Category names and other relevant attributes. In another module we showed you how to efficiently query Categories.

5. Transform and load

• With all the necessary data gathered and enriched, transform it into the format required by the third-party search engine and load it.
• Here you may need to consider things like:
• Will a Product used in multiple Stores be created as a single or multiple index?
• Which fields are absolutely necessary for the third‑party system and which field can I remove?
• Naturally the third‑party search will not have native support for features like Price Hints, how will you support this? One option is to support this logic in your own BFF or front-end, so that your application knows the correct price to display to a given customer

How it works

1. Schedule: Your service "wakes up" on a predefined schedule (for example, every 15 minutes).
2. State Management:
   1. Stateless (Fixed Interval): The service has no memory of past runs. It queries for changes within a fixed, recent time window (for example, a job running every 15 minutes asks for changes from "the last 15 minutes").Note: a single failed or delayed job can create a permanent gap, causing updates to be missed.
   2. Stateful (Persistent Timestamp): The service reads a timestamp from a persistent store that marks the end of the last successful sync. After processing the new changes, it updates the stored timestamp to the current time. This approach guarantees that no data is missed, as it will automatically catch up on all changes since the last successful run, even if the service was down for an extended period.
3. Query for Changes: It queries the relevant endpoints using a where predicate to filter for resources modified since the last sync's timestamp. The lastModifiedAt field is perfect for this.
4. Process and Update: It processes the returned changes, transforms the data as needed (applying tailoring, fetching prices, listing of Products to be disabled etc.), and pushes the updates to the external search engine.

Example Query:

To fetch all Product Projections that have been modified or enabled since the last run:

GET /{{project-key}}/product-projections?where=lastModifiedAt > "2023-10-27T10:00:00.000Z"&staged=false&sort=lastModifiedAt asc&withTotal=false

• Each successive query would update the timestamp to use the time of the last document in the prior response. This is to ensure that we get the oldest lastModifiedAt first and then the newest lastModifiedAt last, this is helpful because it ensures that we process each record at least once and don’t miss any. It is possible that a record gets processed multiple times because it received an update during the sync process. Your application must decide how this should be handled, will you only process the record once or multiple possible time.
• Using sort=lastModifiedAt asc combined with cursor pagination ensures you process changes in chronological order and can handle a large volume of updates within a single polling interval.
• For Product projections you will need to query staged and current separately. This is so that you can get Products that are recently disabled or enabled and/ or have some other change.

You will need to do this query for each entity that you want to check if there is a change for that you want to sync. For example, you may want to check

• If the Selections for a Store have changed
• If the Products or Product Variants on a Selection have changes
• If the Products for a selection is staged or current and/ or has changed values (attributes)
• If the Tailoring is unpublished or published and/ or has changed values

import { apiRoot } from './ctpClient.js';

// The number of results to fetch per API call (page size)
const PAGE_SIZE = 20;

/**
 * Fetches a single page of product projection changes.
 * @param {string} lastTimestamp - The lastModifiedAt from the previous page.
 * @param {string | null} lastId - The ID from the last item of the previous page.
 * @returns {Promise<object>} The API response body.
 */
const fetchNextPageOfChanges = async (lastTimestamp, lastId) => {
  let whereClause;

  // Use an if/else block to construct the WHERE clause.
  if (lastId) {
    // This is for page 2 and onwards. We have a previous product's ID.
    // We query for anything modified AFTER the last timestamp,
    // OR anything modified at the EXACT same timestamp but with a greater ID.
    whereClause = `(lastModifiedAt = "${lastTimestamp}" and id > "${lastId}") or (lastModifiedAt > "${lastTimestamp}")`;
  } else {
    // This is for the very first page of the sync, where lastId is null.
    // The query is simpler.
    whereClause = `lastModifiedAt > "${lastTimestamp}"`;
  }

  const response = await apiRoot
    .productProjections()
    .get({
      queryArgs: {
        where: whereClause,
        staged: false, // Only query for current (published) projections
        sort: ['lastModifiedAt asc', 'id asc'], // Sort by timestamp, then ID
        limit: PAGE_SIZE,
        withTotal: false,
      },
    })
    .execute();

  return response.body;
};

/**
 * Main function to run the delta sync process.
 */
const runDeltaSync = async () => {
  try {
    // 1. Define the starting point for the sync
    const oneHourAgo = new Date(
      new Date().getTime() - 60 * 60 * 1000
    ).toISOString();
    console.log(`--- Starting sync for changes since ${oneHourAgo} ---`);

    // 2. Initialize state variables for pagination
    let lastSyncTimestamp = oneHourAgo;
    let lastId = null;
    let hasMoreResults = true;
    let totalProductsProcessed = 0;
    let page = 1;

    // 3. Loop to fetch all pages of results
    while (hasMoreResults) {
      console.log(`\nFetching page ${page}...`);

      const pageData = await fetchNextPageOfChanges(lastSyncTimestamp, lastId);
      const results = pageData.results;

      if (results.length > 0) {
        console.log(`Found ${results.length} products on this page.`);
        totalProductsProcessed += results.length;

        // Process each product
        results.forEach(product => {
          console.log(
            `  -> Processing ID: ${product.id}, Modified: ${product.lastModifiedAt}`
          );
        });

        // 4. Update the "cursor" with the details of the LAST item in the page
        const lastProduct = results[results.length - 1];
        lastSyncTimestamp = lastProduct.lastModifiedAt;
        lastId = lastProduct.id;

        // If we received fewer results than we asked for, we're on the last page.
        if (results.length < PAGE_SIZE) {
          hasMoreResults = false;
        }
      } else {
        // No more results found, exit the loop.
        hasMoreResults = false;
      }
      page++;
    }

    console.log('\n--- Sync Complete ---');
    console.log(`Total products processed: ${totalProductsProcessed}`);
    console.log(
      `For the next run, start with timestamp: "${lastSyncTimestamp}" and ID: "${lastId}"`
    );
  } catch (error) {
    console.error('An error occurred during the sync process:', error);
  }
};

// Run the sync
runDeltaSync();

How it works

The process is straightforward:

1. Subscribe: You create a Subscription in your commercetools Project, specifying which events you care about (for example, ProductPublished). You configure this Subscription to send a notification message to a destination, like a message queue (for example, Google Pub/Sub, AWS SQS).
2. Trigger: A merchant publishes a new Product in the Merchant Center.
3. Notify: Composable Commerce instantly sends a message to your configured queue. This message acts as a signal, containing a reference to the resource that changed.
4. Process: A service you control is triggered by the new message in the queue.
5. Update: Your service uses the information in the message to make a targeted API call back to commercetools, fetching the full, up-to-date Product data. It then transforms this data into the required format and pushes it to your search engine index.

Choosing the right messages for your search index

For a lean and efficient search integration, you only need to subscribe to two core messages:

1. ProductPublished
2. ProductUnpublished

Handling new and updated Products with ProductPublished

• How it Works: In commercetools, any change to a Product — whether it's a brand new item or just a simple price adjustment, a new image, or an updated description — first creates a staged version. These staged changes are not visible to customers until you explicitly publish them. The act of publishing triggers the ProductPublished message.
• Why it's Efficient: This single message is your perfect trigger. It tells you, "Something is now live for the customer." It doesn't matter if the Product was brand new or just updated; this is the moment you need to add or update the entry in your search index.

Handling Product removals with ProductUnpublished

When a Product needs to be removed from your Store, it must first be unpublished. This action is the ideal signal for removing it from the search index.

• How it Works: A Product cannot be deleted while it is in a published state. The first step is always to unpublish it, which fires the ProductUnpublished message.
• Why it's Sufficient: Because unpublishing is a mandatory prerequisite for deletion, this message serves as the perfect, timely trigger to remove the item from your search index. Listening for a separate ProductDeleted message would be redundant for this purpose.

Configure Composable Commerce Project

Now that the Google Cloud setup is complete, we can move on to configuring the Composable Commerce Project before writing the listener code that will poll our Pub/Sub queue and react to incoming messages.

1. Request method: POST
2. Endpoint: Subscriptions
3. Command: Create Subscription
4. Add the following body and then click Send:

{
  "destination": {
    "type": "GoogleCloudPubSub",
    "topic": "ct-product-sync", //the value we provided in step 3
    "projectId": "ct-search-sync" //the value we provided in step 1
  },
  "messages": [
    {
      "resourceTypeId": "product",
      "types": [
// the two Message Types we want to subscribe to
        "ProductPublished",
        "ProductUnpublished"
      ]
    }
  ],
  "key": "product-search-sync" //the key to identify our Subscription by
}

Alternatively, if you’re managing your infrastructure as code, you can also configure this Subscription using Terraform with the commercetools provider. This makes the setup more repeatable and easier to maintain in the long run.

If the Subscription was created successfully, it means both our Pub/Sub topic and Composable Commerce Project are configured correctly.

Remember: Composable Commerce always sends a test message when creating a Subscription. If that message can’t be delivered, the Subscription won’t be created.

Prepare the listener application

Now, let's focus on developing our listener application. The code below can be re-used depending on your needs: you could have this running in the cloud, or on a Virtual Machine, or even locally.

To begin, ensure that all necessary dependencies for the Google Pub/Sub integration are installed. For npm, execute the following command in your CLI / terminal:

npm install @google-cloud/pubsub

If instead you are using yarn, execute the following command in your CLI/ terminal:

yarn add @google-cloud/pubsub

With that done, we can write our code:

import { PubSub } from "@google-cloud/pubsub";

//...

async updateSearchIndex(product: ProductProjection) {
    console.log(`Updated/added product in search engine: ${product.id}`);
    // TODO: Format and send to search index
  }

async removeFromSearchIndex(productId: string) {
    console.log(`Removed product from search engine: ${productId}`);
    // TODO: Remove by objectID from search index
  }

async fetchMessages(): Promise<void> {
    const subscriptionName = "ct-product-consumer"; //the Subscription ID value we provided in step 4

    // Initialize the Pub/Sub client with explicit credentials
    const pubSubClient = new PubSub({
      keyFilename: "./ctapisubscription.json", // Path to the location of your service account JSON key
      projectId: "ct-search-sync", // the GCP project ID that we set in step 1
    });

    const subscription = pubSubClient.subscription(subscriptionName);

    // Start listening to incoming messages from the subscription
    subscription.on("message", async (message) => {
      try {
        const ctMessage = JSON.parse(message.data.toString());
        const { type, resource } = ctMessage;

        switch (type) {
          case "ProductPublished": {
            // Fetch the published product from a specific store
            const product = await apiRoot
              .inStoreKeyWithStoreKeyValue({ storeKey: "b2c-retail-store" })
              .productProjections()
              .withId({ ID: resource.id })
              .get()
              .execute();

            // Sync the product to the third-party search engine
            await this.updateSearchIndex(product.body);
            break;
          }

          case "ProductUnpublished": {
            // Remove the product from the search index
            await this.removeFromSearchIndex(resource.id);
            break;
          }

          default:
            console.log(` Unhandled message type: ${type}`);
        }

        // Acknowledge successful handling of the message
        message.ack();
      } catch (err) {
        console.error(" Error handling message:", err);

        // Optionally requeue the message for later processing
        message.nack();
      }
    });
    // Error handler
    subscription.on("error", (err) => {
      console.error("Subscription error:", err);
    });
  }

And that’s it! Once you run your listener app, head over to your Project and publish or unpublish a few Products. You should see your application receive and handle the corresponding messages from the Pub/Sub queue.