# ---------------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ---------------------------------------------------------------------------

apiVersion: camel.apache.org/v1alpha1
kind: Kamelet
metadata:
  name: azure-storage-blob-cdc-source
  annotations:
    camel.apache.org/kamelet.support.level: "Preview"
    camel.apache.org/catalog.version: "3.21.0"
    camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHhtbG5zOnhsaW5rPSJodHRwOi8vd3d3LnczLm9yZy8xOTk5L3hsaW5rIiB2aWV3Qm94PSIwIDAgODEgODIiIGZpbGw9IiNmZmYiIGZpbGwtcnVsZT0iZXZlbm9kZCIgc3Ryb2tlPSIjMDAwIiBzdHJva2UtbGluZWNhcD0icm91bmQiIHN0cm9rZS1saW5lam9pbj0icm91bmQiPjx1c2UgeGxpbms6aHJlZj0iI0EiIHg9Ii41IiB5PSIuNSIvPjxzeW1ib2wgaWQ9IkEiIG92ZXJmbG93PSJ2aXNpYmxlIj48cGF0aCBkPSJNMS4zMzMgMEMuNTMzIDAgMCAuNTMzIDAgMS4zMzN2MTZjMCAuOC41MzMgMS4zMzMgMS4zMzMgMS4zMzNoOGMuOCAwIDEuMzMzLS41MzMgMS4zMzMtMS4zMzN2LTYuNjY2aDU4LjY2N3Y2LjY2N2MwIC44LjUzMyAxLjMzMyAxLjYgMS4zMzNoNy43MzNjLjggMCAxLjMzMy0uNTMzIDEuMzMzLTEuMzMzdi04LThDODAgLjUzMyA3OS40NjcgMCA3OC42NjcgMGgwSDEuMzMzem03Ny4zMzQgODBjLjggMCAxLjMzMy0uNTMzIDEuMzMzLTEuMzMzVjYyLjkzNGMwLS44LS41MzMtMS4zMzMtMS4zMzMtMS4zMzNoLTcuNzMzYy0uOCAwLTEuMzMzLjUzMy0xLjMzMyAxLjMzM3Y2LjRIMTAuNjY3di02LjY2N2MwLS44LS41MzMtMS4zMzMtMS42LTEuMzMzSDEuMzMzYy0uOCAwLTEuMzMzLjUzMy0xLjMzMyAxLjZ2MTUuNzMzQzAgNzkuNDY3LjUzMyA4MCAxLjMzMyA4MGg3Ny4zMzR6IiBmaWxsPSIjMDA3MmM2IiBzdHJva2U9Im5vbmUiLz48cGF0aCBkPSJNMjkuNTE5IDM2LjQ0N2wzLjQ1Mi0zLjQ1MiAyLjU1NiAyLjU1Ni0zLjQ1MiAzLjQ1MXptNS4wMDctNS4wMDhsMy40NTYtMy40NTYgMi41NiAyLjU2TDM3LjA4NiAzNHptNy41Ny0yLjQ2M2wtMi41Ni0yLjU0NCAzLjQyNC0zLjQyNCAyLjU2IDIuNTQ0ek0zOS41MzUgNTIuNWwyLjU2MS0yLjU2IDMuNDI0IDMuNDI2LTIuNTYxIDIuNTZ6bS0xLjU1MS0xLjU3MmwtMy40NTYtMy40NTYgMi41Ni0yLjU0NCAzLjQ1NiAzLjQ1NnptLTUuMDA4LTUuMDA4bC0zLjQ1Ni0zLjQ1NiAyLjU2LTIuNTQ0IDMuNDQgMy40NHoiIHN0cm9rZT0ibm9uZSIgZmlsbD0iI2I4ZDQzMiIvPjxwYXRoIGQ9Ik01Ny44MSAyNC4zOTVhNC40NCA0LjQ0IDAgMCAxLTMuMTU4IDEuMzA5IDQuNDMgNC40MyAwIDAgMS0zLjE1OC0xLjMwOWwtNi4wNTMtNi4wNTN2MTIuMzA2aDE4LjQyNFYxOC4zNDJsLTYuMDU0IDYuMDUzem0tMy4xNjEuNDA5YTMuNTQgMy41NCAwIDAgMCAyLjUyMy0xLjA0NWw2LjY5LTYuNjl2LS4zNTVINDUuNDM4di4zNTVsNi42OSA2LjY5YTMuNTQgMy41NCAwIDAgMCAyLjUyMiAxLjA0NXptMy4xNjEgMzEuMTVhNC40NCA0LjQ0IDAgMCAxLTMuMTU4IDEuMzA5IDQuNDMgNC40MyAwIDAgMS0zLjE1OC0xLjMwOWwtNi4wNTMtNi4wNTN2MTIuMzA2aDE4LjQyNFY0OS45MDFsLTYuMDU0IDYuMDUzem0tMy4xNjEuNDFhMy41NCAzLjU0IDAgMCAwIDIuNTIzLTEuMDQ1bDYuNjktNi42ODl2LS4zNTVINDUuNDM4di4zNTVsNi42OSA2LjY4OWEzLjU0IDMuNTQgMCAwIDAgMi41MjIgMS4wNDV6TTE5LjcxOCA0MC41OTljLTEuMzQ0IDEuMzQ0LTMuMTMzIDIuMDg1LTUuMDM1IDIuMDg1cy0zLjY5LS43NDEtNS4wMzUtMi4wODVMMCAzMC45NDl2MTkuNjEzaDI5LjM2NlYzMC45NDlsLTkuNjQ4IDkuNjV6bS01LjAzNS42NTJhNS42NSA1LjY1IDAgMCAwIDQuMDIyLTEuNjY2bDEwLjY2MS0xMC42NjF2LS41NjVIMHYuNTY2bDEwLjY2MSAxMC42NjFhNS42NiA1LjY2IDAgMCAwIDQuMDIyIDEuNjY0eiIgc3Ryb2tlPSJub25lIiBmaWxsPSIjNTliNGQ5Ii8+PC9zeW1ib2w+PC9zdmc+"
    camel.apache.org/provider: "Apache Software Foundation"
    camel.apache.org/kamelet.group: "Azure Servicebus"
    camel.apache.org/kamelet.namespace: "Azure"
  labels:
    camel.apache.org/kamelet.type: "source"
spec:
  definition:
    title: Azure Storage Blob CDC Source
    description: |-
      Receive data from Azure Service Bus subscribed to Azure Eventgrid reporting events related to a Azure Storage Blob account.

      Connection string is the basic method for authenticating to the Azure Servicebus Queue.

      To use this Kamelet you'll need to set up Events on your Azure Storage Blob account and select as an endpoint an Azure Servicebus Queue.
    required:
      - topicOrQueueName
      - connectionString
      - accountName
      - containerName
      - accessKey
    type: object
    properties:
      topicOrQueueName:
        title: Topic Or Queue Name
        description: Topic Or Queue Name for the Azure Servicebus instance
        type: string
      connectionString:
        title: Connection String
        description: Connection String for Azure Servicebus instance
        type: string
        format: password
        x-descriptors:
        - urn:alm:descriptor:com.tectonic.ui:password
        - urn:camel:group:credentials
      serviceBusReceiveMode:
        title: Servicebus Receive Mode
        description: Sets the receive mode for the receiver.
        type: string
        default: RECEIVE_AND_DELETE
        enum: ["RECEIVE_AND_DELETE", "PEEK_LOCK"]
      subscriptionName:
        title: Subscription Name
        description: Sets the name of the subscription in the topic to listen to. This parameter is mandatory in case of topic.
        type: string
      accountName:
        title: Account Name
        description: The Azure Storage Blob account name.
        type: string
        x-descriptors:
        - urn:camel:group:credentials
      containerName:
        title: Container Name
        description: The Azure Storage Blob container name.
        type: string
      accessKey:
        title: Access Key
        description: The Azure Storage Blob access key.
        type: string
        format: password
        x-descriptors:
        - urn:alm:descriptor:com.tectonic.ui:password
        - urn:camel:group:credentials
      credentialType:
        title: Credential Type
        description: Determines the credential strategy to adopt.
        type: string
        default: SHARED_ACCOUNT_KEY
        enum: ["SHARED_ACCOUNT_KEY", "SHARED_KEY_CREDENTIAL", "AZURE_IDENTITY"]
      getBlob:
        title: Get Object in Container
        description: >-
          If getBlob is enabled, then the file created in the container will be
          get and returned as body, if not only the event will be returned as body.
        type: boolean
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: false
  types:
    out:
      mediaType: application/octet-stream
  dependencies:
    - "camel:azure-servicebus"
    - "camel:azure-storage-blob"
    - "camel:kamelet"
    - "camel:core"
    - 'camel:jsonpath'
    - 'camel:jackson'
  template:
    from:
      uri: "azure-servicebus:{{topicOrQueueName}}"
      parameters:
        connectionString: "{{connectionString}}"
        serviceBusReceiveMode: "{{serviceBusReceiveMode}}"
        subscriptionName: "{{?subscriptionName}}"
      steps:
        - choice:
            precondition: true
            when:
              - simple: '${properties:getBlob:true}'
                steps:
                  - set-body:
                      simple: ${body.toString()}
                  - unmarshal:
                      json:
                        library: Jackson
                        unmarshalType: com.fasterxml.jackson.databind.JsonNode
                  - set-property:
                      name: azure-storage-blob-event-type
                      jsonpath: $.eventType
                  - set-property:
                      name: azure-storage-blob-subject
                      jsonpath: $.subject
                  - set-property:
                      name: azure-storage-blob-blob-name
                      simple: '${exchangeProperty.azure-storage-blob-subject.substring(${exchangeProperty.azure-storage-blob-subject.lastIndexOf("/")}++)}'
                  - log: "${exchangeProperty.azure-storage-blob-event-type} - ${exchangeProperty.azure-storage-blob-subject} - ${exchangeProperty.azure-storage-blob-blob-name}"
                  - choice:
                      when:
                        - simple: '${exchangeProperty.azure-storage-blob-event-type} == "Microsoft.Storage.BlobCreated" && ${exchangeProperty.azure-storage-blob-subject} contains "{{containerName}}"'
                          steps:
                            - toD: >-
                                azure-storage-blob:{{accountName}}/{{containerName}}?accessKey=RAW({{accessKey}})&operation=getBlob&blobName=${exchangeProperty.azure-storage-blob-blob-name}&credentialType={{credentialType}}
        - to: 'kamelet:sink'