id: azure_eventhubs_realtime_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: Hello there! I received {{ trigger.body }} from Azure EventHubs!

triggers:
  - id: read_from_eventhub
    type: io.kestra.plugin.azure.eventhubs.RealtimeTrigger
    eventHubName: my_eventhub
    namespace: my_eventhub_namespace
    connectionString: "{{ secret('EVENTHUBS_CONNECTION') }}"
    bodyDeserializer: JSON
    consumerGroup: "$Default"
    checkpointStoreProperties:
      containerName: kestra
      connectionString: "{{ secret('BLOB_CONNECTION') }}"

    id: eventhubs_realtime_trigger
    namespace: company.team

    tasks:
      - id: insert_into_storagetable
        type: io.kestra.plugin.azure.storage.table.Bulk
        endpoint: https://yourstorageaccount.blob.core.windows.net
        connectionString: "{{ secret('STORAGETABLE_CONNECTION') }}"
        table: orders
        from:
          - partitionKey: order_id
            rowKey: "{{ trigger.body | jq('.order_id') | first }}"
            properties:
              customer_name: "{{ trigger.body | jq('.customer_name') | first }}"
              customer_email: "{{ trigger.body | jq('.customer_email') | first }}"
              product_id: "{{ trigger.body | jq('.product_id') | first }}"
              price: "{{ trigger.body | jq('.price') | first }}"
              quantity: "{{ trigger.body | jq('.quantity') | first }}"
              total: "{{ trigger.body | jq('.total') | first }}"

    triggers:
      - id: realtime_trigger
        type: io.kestra.plugin.azure.eventhubs.RealtimeTrigger
        eventHubName: orders
        namespace: kestra
        connectionString: "{{ secret('EVENTHUBS_CONNECTION') }}"
        bodyDeserializer: JSON
        consumerGroup: $Default
        checkpointStoreProperties:
          containerName: kestra
          connectionString: "{{ secret('BLOB_CONNECTION') }}"