apiVersion: camel.apache.org/v1alpha1
kind: Kamelet
metadata:
  name: aws-s3-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,[encoded data, 4228 characters]
    camel.apache.org/provider: Apache Software Foundation
    camel.apache.org/kamelet.group: AWS S3 CDC
    camel.apache.org/kamelet.namespace: "AWS"
    camel.apache.org/keda.type: aws-s3-cdc-queue
  labels:
    camel.apache.org/kamelet.type: source
spec:
  definition:
    title: AWS S3 CDC Source
    description: >-
      Receive data from AWS SQS subscribed to Eventbridge Bus reporting events related to an S3 bucket or multiple buckets.

      Access Key/Secret Key are the basic method for authenticating to the AWS
      SQS Service.

      To use this Kamelet you'll need to set up Eventbridge on your bucket and subscribe Eventbridge bus to an SQS Queue.
      
      For doing this you'll need to enable Evenbridge notification on your bucket and creating a rule on Eventbridge console related to all the events on S3 bucket and pointing to the SQS Queue specified as parameter in this Kamelet.
    required:
      - accessKey
      - secretKey
      - queueNameOrArn
      - region
    type: object
    properties:
      queueNameOrArn:
        title: Queue Name
        description: The SQS Queue Name or ARN
        type: string
      deleteAfterRead:
        title: Auto-delete Messages
        description: Delete messages after consuming them
        type: boolean
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: true
      accessKey:
        title: Access Key
        description: The access key obtained from AWS.
        type: string
        format: password
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:password'
          - 'urn:camel:group:credentials'
          - 'urn:keda:authentication:awsAccessKeyID'
          - 'urn:keda:required'
      secretKey:
        title: Secret Key
        description: The secret key obtained from AWS.
        type: string
        format: password
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:password'
          - 'urn:camel:group:credentials'
          - 'urn:keda:authentication:awsSecretAccessKey'
          - 'urn:keda:required'
      region:
        title: AWS Region
        description: The AWS region to access.
        type: string
        x-descriptors:
          - 'urn:keda:metadata:awsRegion'
          - 'urn:keda:required'
        enum:
          - ap-south-1
          - eu-south-1
          - us-gov-east-1
          - me-central-1
          - ca-central-1
          - eu-central-1
          - us-iso-west-1
          - us-west-1
          - us-west-2
          - af-south-1
          - eu-north-1
          - eu-west-3
          - eu-west-2
          - eu-west-1
          - ap-northeast-3
          - ap-northeast-2
          - ap-northeast-1
          - me-south-1
          - sa-east-1
          - ap-east-1
          - cn-north-1
          - us-gov-west-1
          - ap-southeast-1
          - ap-southeast-2
          - us-iso-east-1
          - ap-southeast-3
          - us-east-1
          - us-east-2
          - cn-northwest-1
          - us-isob-east-1
          - aws-global
          - aws-cn-global
          - aws-us-gov-global
          - aws-iso-global
          - aws-iso-b-global
      autoCreateQueue:
        title: Autocreate Queue
        description: Setting the autocreation of the SQS queue.
        type: boolean
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: false
      amazonAWSHost:
        title: AWS Host
        description: The hostname of the Amazon AWS cloud.
        type: string
        default: amazonaws.com
      protocol:
        title: Protocol
        description: The underlying protocol used to communicate with SQS
        type: string
        example: http or https
        default: https
      queueURL:
        title: Queue URL
        description: The full SQS Queue URL (required if using KEDA)
        type: string
        x-descriptors:
          - 'urn:keda:metadata:queueURL'
          - 'urn:keda:required'
      uriEndpointOverride:
        title: Overwrite Endpoint URI
        description: >-
          The overriding endpoint URI. To use this option, you must also select
          the `overrideEndpoint` option.
        type: string
      overrideEndpoint:
        title: Endpoint Overwrite
        description: >-
          Select this option to override the endpoint URI. To use this option,
          you must also provide a URI for the `uriEndpointOverride` option.
        type: boolean
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: false
      delay:
        title: Delay
        description: The number of milliseconds before the next poll of the selected stream
        type: integer
        default: 500
      greedy:
        title: Greedy Scheduler
        description: >-
          If greedy is enabled, then the polling will happen immediately again,
          if the previous run polled 1 or more messages.
        type: boolean
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: false
      getObject:
        title: Greedy Object in Bucket
        description: >-
          If getObject is enabled, then the file created in the bucket will be
          get and returned as body, if not only the event will returned as body.
        type: boolean
        x-descriptors:
          - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: false
  dependencies:
    - 'camel:core'
    - 'camel:aws2-sqs'
    - 'camel:aws2-s3'
    - 'camel:jsonpath'
    - 'camel:kamelet'
    - 'camel:jackson'
  template:
    from:
      uri: 'aws2-sqs:{{queueNameOrArn}}'
      parameters:
        autoCreateQueue: '{{autoCreateQueue}}'
        secretKey: '{{?secretKey}}'
        accessKey: '{{?accessKey}}'
        region: '{{region}}'
        deleteAfterRead: '{{deleteAfterRead}}'
        amazonAWSHost: '{{?amazonAWSHost}}'
        protocol: '{{?protocol}}'
        uriEndpointOverride: '{{?uriEndpointOverride}}'
        overrideEndpoint: '{{overrideEndpoint}}'
        delay: '{{delay}}'
        greedy: '{{greedy}}'
      steps:
        - choice:
            precondition: true
            when:
              - simple: '${properties:getObject:true}'
                steps:
                  - unmarshal:
                      json:
                        library: Jackson
                        unmarshalType: com.fasterxml.jackson.databind.JsonNode
                  - set-property:
                      name: s3-event-name
                      jsonpath: $.detail.reason
                  - choice:
                      when:
                        - simple: '${exchangeProperty.s3-event-name} == "PutObject"'
                          steps:
                            - set-property:
                                name: aws-s3-name
                                jsonpath: $.detail.object.key
                            - set-property:
                                name: aws-s3-bucket
                                jsonpath: $.detail.bucket.name
                            - toD: >-
                                aws2-s3:${exchangeProperty.aws-s3-bucket}?accessKey={{accessKey}}&secretKey={{secretKey}}&region={{region}}&operation=getObject&keyName=${exchangeProperty.aws-s3-name}
        - to: 'kamelet:sink'