Middleware

Middleware serves as an intermediary layer that processes incoming changes from data sources before they are passed to the query engine. Middleware components are modular and can be stacked or combined in a pipeline to process incoming changes sequentially.

Its primary role is transformation, modifying or enriching the data as needed, such as normalizing values, applying mappings, or adding computed fields.

The configuration settings in the spec.sources.middleware section of the Continuous Query resource definition hold the individual middleware configurations that can be used in a pipeline for a given source.

Each middleware definition requires a name, which is referenced in by spec.sources.subscriptions.pipeline, a kind, which defines which middleware implementation to use and and properties required by that specific implementation.

Middleware components

Unwind

The unwind middleware component can be used to unwind an array of values that is nested inside the properties of a Node or Relation. Unwinding the array will create new top level elements in the graph that can be referenced as such using Cypher.

The configuration for the unwind component are as follows

In addition to the kind and name properties, any additional properties on this configuration object will be a dictionary of unwind configurations, where the key is the label of the incoming node.

The configuration properties for unwinding an element are as follows

Example

For example, imagine you had a source that produced nodes with the label of Vehicle and the properties looked like this:

{
    "id": "vehicle-1",
    "tires": [
        {
            "position": "Front-Left",
            "pressure": 250
        },
        {
            "position": "Front-Right",
            "pressure": 258
        },
        {
            "position": "Rear-Left",
            "pressure": 243
        },
        {
            "position": "Rear-Right",
            "pressure": 252
        }
    ]
}

We could unwind the tires array into top level nodes with the following metadata and corresponding query.

apiVersion: v1
kind: ContinuousQuery
name: query
spec:
  sources:    
    subscriptions:
      - id: source
        nodes:
          - sourceLabel: Vehicle
        pipeline:
          - extract-tires
    middleware:
      - name: extract-tires
        kind: unwind
        Vehicle:
          - selector: $.tires[*]
            label: Tire
            key: $.position
            relation: HAS
  query: >
    MATCH
        (v:Vehicle)-[:HAS]->(t:Tire)
    RETURN
        v.id,
        t.position,
        t.pressure    

Map

The map middleware component can be used to remap an incoming insert/update/delete from a source to a different insert/update/delete for another element.

The configuration properties for the map component are as follows

The configuration for mapping an element is as follows

Example

For example, if you had a source that was an append only log of sensor readings, but your query is only ever interested in the latest value.

{
    "id": "log-1",
    "sensorId": "thermostat-1",
    "value": 25
}

{
    "id": "log-2",
    "sensorId": "thermostat-1",
    "value": 27
}

{
    "id": "log-3",
    "sensorId": "thermostat-1",
    "value": 28
}

You can remap the inserts of the SensorLog to updates of a Sensor, so if you get 3 updates for a given sensor over time, you only need to index the last one, rather than the entire history. In this configuration, the root of the SensorLog payload is selected to update a Sensor with the ID equal to the sensorId field in the incoming change payload. So instead of 3 SensorLog nodes, we now have 1 Sensor node with the current value.

apiVersion: v1
kind: ContinuousQuery
name: query
spec:
  sources:    
    subscriptions:
      - id: source
        nodes:
          - sourceLabel: SensorLog
        pipeline:
          - extract-latest
    middleware:
      - kind: map
        name: extract-latest
        SensorLog:        
          insert:
            - selector: $
              op: Update
              label: Sensor
              id: $.sensorId
              properties:
                sensorId: $.sensorId
                currentValue: $.value

  query: >
    MATCH
        (s:Sensor)
    RETURN
        s.sensorId,
        s.currentValue        

Promote

The promote middleware processes SourceChange events (Insert and Update) to copy values from deep-nested locations inside an element’s properties map to new top-level properties. Selection is performed with JSONPath expressions, and each promoted value is written under an explicit target_name. This is useful for flattening complex structures or making frequently accessed data more readily available.

The configuration for the promote component is as follows:

The Mapping Object within the config.mappings array has the following properties:

Example

Here’s an example of how to configure the promote middleware to extract user and order data to top-level properties. This configuration will attempt to promote several fields; if a target property already exists, it will be skipped, and if a JSONPath expression fails to resolve, that specific mapping will be skipped.

spec:
  sources:
    middleware:
      - name: promote_user_and_order_data
        kind: promote
        config:
          mappings:
            - path: "$.user.id"
              target_name: "userId"
            - path: "$.user.location.city"
              target_name: "city"
            - path: "$.order.total"
              target_name: "orderTotal"
            - path: "$.metadata" # Promoting an entire object
              target_name: "meta"
          on_conflict: skip   # Keep existing values if 'userId', 'city', etc. already exist
          on_error:    skip   # Skip mappings that error (e.g., if '$.order.total' doesn't exist)

For instance, if an incoming node has properties like this:

{
    "user": {
        "id": "user123",
        "location": {
            "city": "New York"
        }
    },
    "order": {
        "total": 100.50
    },
    "metadata": { "source": "api", "version": "1.1" }
}

After processing with the promote_user_and_order_data middleware configured above, the node’s properties would be transformed to:

{
    "user": {
        "id": "user123",
        "location": {
            "city": "New York"
        }
    },
    "order": {
        "total": 100.50
    },
    "metadata": { "source": "api", "version": "1.1" },
    "userId": "user123",
    "city": "New York",
    "orderTotal": 100.50,
    "meta": { "source": "api", "version": "1.1" }
}

The userId, city, orderTotal, and meta fields are now available as top-level properties.

ParseJson

The parse_json middleware processes SourceChange events (specifically Insert and Update) to parse a string property containing a JSON document into a structured ElementValue (Object or List). This is useful when a data source provides JSON data embedded within a string field, which needs to be accessible as a structured object or array for querying.

The configuration for the parse_json component is as follows:

Example

Here’s an example of how to configure the parse_json middleware to parse a JSON string from the raw_event_json property and store the resulting structured object in a new property named event_details. If an error occurs during parsing, the change will be skipped.

spec:
  sources:
    middleware:
      - name: parse_event_data
        kind: parse_json
        config:
          target_property: "raw_event_json"
          output_property: "event_details"
          on_error: "skip"

For example, if an incoming node has properties like:

{
    "id": "event001",
    "timestamp": "2025-06-01T12:00:00Z",
    "raw_event_json": "{\"user\": \"alice\", \"action\": \"login\", \"details\": {\"ip\": \"192.168.1.100\"}}"
}

After processing with the parse_event_data middleware, the node’s properties would be transformed to:

{
    "id": "event001",
    "timestamp": "2025-06-01T12:00:00Z",
    "raw_event_json": "{\"user\": \"alice\", \"action\": \"login\", \"details\": {\"ip\": \"192.168.1.100\"}}",
    "event_details": {
        "user": "alice",
        "action": "login",
        "details": {
            "ip": "192.168.1.100"
        }
    }
}

The event_details property now contains the parsed JSON object, which can be queried directly.

Decoder

The decoder middleware processes SourceChange events (specifically Insert and Update) to decode a string value found in a specified property of an Element. It supports various common encoding formats such as Base64, Hexadecimal, URL encoding, and JSON string escapes. This is useful when data from a source is encoded for transmission or storage.

The configuration for the decoder component is as follows:

Encoding Types Supported:

• base64: Standard Base64 encoding (RFC 4648).
• base64url: URL-safe Base64 encoding (RFC 4648 §5), without padding.
• hex: Hexadecimal encoding (e.g., 48656c6c6f).
• url: Percent-encoding (e.g., Hello%20World).
• json_escape: Decodes JSON string escape sequences (e.g., \", \\, \n, \uXXXX). Assumes the input is the content of a JSON string literal, not the literal itself including quotes.

Example

Here’s an example of how to configure the decoder middleware to decode a Base64 encoded string. The string is located in the raw_user_payload property, may be surrounded by quotes which should be stripped, and the decoded result will be stored in a new user_data property. Errors will be skipped.

spec:
  sources:
    middleware:
      - name: decode_user_data
        kind: decoder
        config:
          encoding_type: "base64"
          target_property: "raw_user_payload"
          output_property: "user_data"
          strip_quotes: true
          on_error: "skip"

For example, if an incoming node has properties like:

{
    "message_id": "msg123",
    "raw_user_payload": "\"SGVsbG8gV29ybGQh\"" 
}

(The string SGVsbG8gV29ybGQh is “Hello World!” encoded in Base64.)

After processing with the decode_user_data middleware:

1. strip_quotes: true removes the surrounding double quotes from "SGVsbG8gV29ybGQh" to yield SGVsbG8gV29ybGQh.
2. This resulting string is then Base64 decoded to Hello World!.

The node’s properties would be transformed to:

{
    "message_id": "msg123",
    "raw_user_payload": "\"SGVsbG8gV29ybGQh\"",
    "user_data": "Hello World!"
}

The user_data property now contains the decoded string “Hello World!”.