Aggregate Inventory Sync - commercetools Connector
Essential knowledge
Intended Audience:
Technical User
Author:
Fluent Commerce
Changed on:
17 June 2026
Overview
The Aggregate Inventory Sync module enables product inventory synchronization from Fluent Order Management to commercetools at an aggregate level. Under this architecture, Fluent Order Management acts as the master single source of truth for all inventory data. When aggregate quantities change, the connector distributes those updates to matching storefront network configurations.Key points
- Core Function & Outcome: You will learn how the module runs scheduled batch jobs to fetch available product quantities from Fluent Order Management and update aggregate inventory balances within commercetools channels.
- Single Source of Truth: Fluent Order Management serves as the authoritative master environment for tracking, computing, and distributing all aggregate stock positions down to the storefront.
- Scheduled Batch Orchestration: The inventory synchronization pipeline runs via a dedicated batch job handler on a configurable frequency. Out-of-the-box orchestration leverages AWS Event Bridge scheduled triggers deployed via cloud formation templates.
- Custom Handler Adaptability: Technical partners can cleanly alter inventory processing logic or add custom validation metrics by overriding default routing handlers within the standard configuration files.
Data Pipeline Execution
1. Ingestion (Webhook/Batch Job)
The inventory synchronization pipeline runs via a dedicated batch job handler on a configurable frequency. Out-of-the-box orchestration leverages external scheduling systems (such as AWS Event Bridge scheduled triggers) to send a standardized HTTP request to the connector framework:HTTP`PUT https://<domain>/api/v1/fluent-connect/scheduler/add/<fluent-account>/<fluent-retailer>/batch-inventory-sync`This trigger routes execution variables into an internal processing network. The bulk tracking data is funneled into a specialized Batch Queue designed specifically to handle high-throughput aggregate synchronization tasks safely.
2. Processing & Payload Extraction
The internal Job Router evaluates the explicit name of the incoming job payload and forwards it directly to the assigned Inventory Batch Handler class.The handler uses the pre-calculated date ranges provided by the job context to run in delta mode, only fetching items that have changed since the last execution. It dispatches an outbound GraphQL query (`GetAggregateVirtualPositions`) to extract clean availability quantities from Fluent Order Management, filtering records by virtual catalogs, status attributes, and timestamps.3. Target State Mapping (commercetools Mutation/Fields)
Once the aggregate balances are extracted from Fluent, the connector updates the corresponding storefront channels:- Channel Key Matching: The module reviews settings defined under the profile name
`fc.connect.commercetools.batch.batch-inventory-sync`to correlate your target storefront channel key (`ctChannelKey`) with the authoritative Fluent virtual catalog reference (`fluentAggregateVirtualCatalogueRef`).
- Inventory Balance Upserts: The connector runs upsert operations to create or update inventory entries within the designated commercetools channel, resetting available-to-sell item counts to align perfectly with master values.
Data Storage & Schema Mapping
Batch Inventory Synchronization Setting Profile
The following schema profile details the structure of an active inventory sync setting and its corresponding runtime parameters:1{
2 "previousEndDate": "2022-10-20T04:12:08.812570167",
3 "props": {
4 "ctChannelKey": "ct-fluent-channel",
5 "fluentAggregateVirtualCatalogueRef": "ATS:1"
6 },
7 "lastRun": {
8 "param": {
9 "start": "2022-10-19T04:12:08.491444877",
10 "end": "2022-10-20T04:12:08.812570167",
11 "props": {
12 "ctChannelKey": "ct-fluent-channel",
13 "fluentAggregateVirtualCatalogueRef": "ATS:1"
14 }
15 },
16 "jobStart": "2022-10-20T04:12:08.716938923",
17 "jobEnd": "2022-10-20T04:12:08.836448368",
18 "status": "SUCCESSFUL"
19 }
20}Developer Extension Points
Refer to the customization guidelines for more information.Overriding Existing Route Handlers
To modify the underlying logic of the Inventory Upsert handler, update the configuration parameters inside`application-connector.yml` under the `routes > name > handler` block:1name: "commercetools.connect.inventory.upsert"
2 handler: "InventoryUpsertHandler"
3 props:
4 payload: "inventory"
5 query-inventories: "ct-inventories.graphql"
6 query-channel: "ct-channel.graphql"
7 mutation-create-inventory: "ct-createinventory.graphql"
8 mutation-update-inventory: "ct-updateinventory.graphql"Custom Message Handler Implementation
Developers can introduce new custom handling components by extending the base class and registering the package under the root SDK path:1@Slf4j
2@Component
3@HandlerInfo(name = "InventoryUpsertHandler", description = "Upsert Inventory at commercetools" )
4public class FluentInventoryUpsertHandler extends BaseMessageHandler {
5 private static final String Aggregate_Inventory = "inventory";
6 // write your logic
7}Fluent Inventory Extraction Query
The batch job dispatches this GraphQL statement to extract computed availability balances from Fluent Order Management:1query GetAggregateVirtualPositions($after: String, $first: Int, $catalogueRef: String!, $status: [String!], $updatedOnFrom: DateTime, $updatedOnTo: DateTime) {
2 virtualPositions(after: $after, first: $first, catalogue: {ref: $catalogueRef}, status: $status, updatedOn: {from: $updatedOnFrom, to: $updatedOnTo}) {
3 virtualPositionEdge: edges {
4 virtualPositionNode: node {
5 quantity
6 productRef
7 }
8 cursor
9 }
10 pageInfo {
11 hasNextPage
12 }
13 }
14}
