Why orchestrate the sync

Why orchestrate the sync

If your services already talk over RabbitMQ, your data changes are already a stream of messages: product updates, inventory changes, new content. Turning that stream into a live Meilisearch index normally means writing and operating yet another consumer service. With Kestra it’s a trigger and a couple of tasks, in declarative YAML. This guide builds it in two stages: a batch consume-and-index flow to see the pieces, then a real-time trigger that indexes each message as it arrives. Because a queue consumer acknowledges messages as it goes, this is incremental by construction, so there’s no separate “first load versus incremental” to manage, just the stream. (This uses the AMQP 0-9-1 protocol, so it works with RabbitMQ and other AMQP brokers.)

Why orchestrate the sync

A hand-written RabbitMQ to Meilisearch consumer means managing connections, acknowledgements, deserialization, batching, retries, restarts, and monitoring, a service to keep alive. Kestra collapses that into a trigger plus tasks, with acknowledgement, retries, and observability handled for you.

Prerequisites

A running Kestra with three plugins (Meilisearch, AMQP, and the transform plugin for reshaping messages), plus a Meilisearch Cloud project and a RabbitMQ broker:

services:
  rabbitmq:
    image: rabbitmq:3-management-alpine
    ports: ["5672:5672", "15672:15672"]

  kestra:
    image: kestra/kestra:latest
    command: server local
    ports: ["8080:8080"]
    environment:
      # your Meilisearch Cloud Default Admin API key, base64-encoded
      SECRET_MEILISEARCH_API_KEY: <base64 of your admin API key>

FROM kestra/kestra:latest
RUN /app/kestra plugins install \
      io.kestra.plugin:plugin-meilisearch:LATEST \
      io.kestra.plugin:plugin-amqp:LATEST \
      io.kestra.plugin:plugin-transform-json:LATEST

Get your Cloud credentials. In the Meilisearch Cloud dashboard, create a project and copy its Project URL (the url in the flows below) and its Default Admin API Key (Settings, then API Keys). Store the key as the MEILISEARCH_API_KEY Kestra secret, and keep broker credentials in secrets too.

The events are JSON messages describing products, for example { “id”: “prod-1”, “name”: “Mechanical Keyboard”, “stock”: 42 }.

Understanding the shape: a batch consume-and-index flow

Before wiring up real-time, here’s a flow you can run on demand. It declares a queue, publishes a few test messages, consumes them, reshapes them, and indexes them:

id: amqp_to_meilisearch
namespace: company.search

variables:
  meilisearch_url: https://ms-xxxxxxxxxxxx-xxxx.meilisearch.io   # your Meilisearch Cloud Project URL
  index: amqp_products
  queue: product-updates

tasks:
  - id: create_queue
    type: io.kestra.plugin.amqp.CreateQueue
    host: rabbitmq
    port: "5672"
    username: guest
    password: guest
    virtualHost: /
    name: "{{ vars.queue }}"
    durability: true

  - id: publish                # stand-in for your real upstream producer
    type: io.kestra.plugin.amqp.Publish
    host: rabbitmq
    port: "5672"
    username: guest
    password: guest
    virtualHost: /
    exchange: ""               # default exchange: routingKey == queue name
    routingKey: "{{ vars.queue }}"
    serdeType: JSON
    from:
      - data: { id: prod-1, name: Mechanical Keyboard, stock: 42 }
      - data: { id: prod-2, name: Wireless Mouse, stock: 130 }
      - data: { id: prod-3, name: 4K Monitor, stock: 7 }

  - id: consume
    type: io.kestra.plugin.amqp.Consume
    host: rabbitmq
    port: "5672"
    username: guest
    password: guest
    virtualHost: /
    queue: "{{ vars.queue }}"
    serdeType: JSON
    maxRecords: 3

  - id: extract_payload
    type: io.kestra.plugin.transform.jsonata.TransformItems
    from: "{{ outputs.consume.uri }}"
    expression: data           # keep only the message payload

  - id: index_documents
    type: io.kestra.plugin.meilisearch.DocumentAdd
    from: "{{ outputs.extract_payload.uri }}"
    index: "{{ vars.index }}"
    url: "{{ vars.meilisearch_url }}"
    key: "{{ secret('MEILISEARCH_API_KEY') }}"

The one non-obvious step is extract_payload. The Consume task writes a full AMQP envelope per message: data (the body), plus headers, contentType, messageId, timestamp, and so on. You don’t want all that in your search index, just the body, so a JSONata TransformItems with expression: data plucks the payload out of each record. (In this demo create_queue and publish set the stage. In production you’d delete them, since your services already produce to the queue.)

The real deal: real-time indexing

Kestra’s AMQP RealtimeTrigger holds a persistent consumer open and starts one execution per message the instant it’s delivered. A published event is searchable in Meilisearch within seconds, with no cron and no polling.

id: amqp_realtime_to_meilisearch
namespace: company.search

variables:
  meilisearch_url: https://ms-xxxxxxxxxxxx-xxxx.meilisearch.io   # your Meilisearch Cloud Project URL
  index: live_products

triggers:
  - id: on_message
    type: io.kestra.plugin.amqp.RealtimeTrigger
    host: rabbitmq
    port: "5672"
    username: guest
    password: guest
    virtualHost: /
    queue: live-products
    serdeType: JSON

tasks:
  - id: write_document
    type: io.kestra.plugin.core.storage.Write
    extension: .ion
    content: "{{ trigger.data | toJson }}"

  - id: index_document
    type: io.kestra.plugin.meilisearch.DocumentAdd
    from: "{{ outputs.write_document.uri }}"
    index: "{{ vars.index }}"
    url: "{{ vars.meilisearch_url }}"
    key: "{{ secret('MEILISEARCH_API_KEY') }}"

Here the message body is available directly as {{ trigger.data }}. The flow writes it to internal storage as an ION document and indexes it. Publish a message to live-products and it shows up in search seconds later. Because DocumentAdd is add-or-replace, this is automatically an upsert stream: publish an updated event for prod-1 and it overwrites the existing document by primary key, exactly right when a queue carries a changelog of your entities.

Handling deletes

Represent deletions as messages and act on them. Publish an explicit delete event, for example { “id”: “prod-1”, “op”: “delete” }, and branch on it inside the per-message execution: route delete events to Meilisearch’s documents/delete-batch endpoint and everything else to DocumentAdd. An If task reading trigger.data does the routing:

triggers:
  - id: on_message
    type: io.kestra.plugin.amqp.RealtimeTrigger
    host: rabbitmq
    port: "5672"
    username: guest
    password: guest
    virtualHost: /
    queue: live-products
    serdeType: JSON

tasks:
  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ trigger.data.op == 'delete' }}"
    then:
      - id: delete_document
        type: io.kestra.plugin.core.http.Request
        uri: "{{ vars.meilisearch_url }}/indexes/{{ vars.index }}/documents/delete-batch"
        method: POST
        contentType: application/json
        body: "[{{ trigger.data.id | toJson }}]"
        headers:
          Authorization: "Bearer {{ secret('MEILISEARCH_API_KEY') }}"
    else:
      - id: write_document
        type: io.kestra.plugin.core.storage.Write
        extension: .ion
        content: "{{ trigger.data | toJson }}"
      - id: index_document
        type: io.kestra.plugin.meilisearch.DocumentAdd
        from: "{{ outputs.write_document.uri }}"
        index: "{{ vars.index }}"
        url: "{{ vars.meilisearch_url }}"
        key: "{{ secret('MEILISEARCH_API_KEY') }}"

The delete branch sends a one-element array ([“prod-1”]) to documents/delete-batch. For the database-side soft-delete pattern (query recently-deleted rows, then delete-batch), see Connect PostgreSQL to Meilisearch with Kestra.

Going to production

Drop the producer tasks. create_queue and publish are only there to generate test data. In production your services produce to the queue and Kestra only consumes.
Real-time versus batch. The RealtimeTrigger gives second-scale freshness, one execution per message. For very high volume, the batch Consume pattern with a larger maxRecords on a schedule is more efficient. Choose freshness or throughput.
Acknowledgement. By default the consumer acks messages as it processes them, so a restart resumes from the queue without reprocessing. Your incrementality comes for free from the broker.
Retries. Add a retry block so a transient Meilisearch error re-attempts the message rather than dropping it. Pair with a dead-letter queue on the broker for poison messages.

Wrap-up

Kestra turns “keep Meilisearch in sync with a RabbitMQ queue” into a trigger and a couple of tasks. The RealtimeTrigger gives you live indexing with acknowledgement handled for you. Add-or-replace semantics make the stream an idempotent upsert feed, and deletes are just another message type. Publish events as you already do, and Kestra keeps search live.

​ Why orchestrate the sync（3）

​ Why orchestrate the sync