huginn/app/models/agents/s3_agent.rb

module Agents
  class S3Agent < Agent
    include FormConfigurable
    include FileHandling

    emits_file_pointer!
    no_bulk_receive!

    default_schedule 'every_1h'

    gem_dependency_check { defined?(Aws::S3) }

    description do
      <<-MD
        The S3Agent can watch a bucket for changes or emit an event for every file in that bucket. When receiving events, it writes the data into a file on S3.

        #{'## Include `aws-sdk-core` in your Gemfile to use this Agent!' if dependencies_missing?}

        `mode` must be present and either `read` or `write`, in `read` mode the agent checks the S3 bucket for changed files, with `write` it writes received events to a file in the bucket.

        ### Universal options

        To use credentials for the `access_key` and `access_key_secret` use the liquid `credential` tag like so `{% credential name-of-credential %}`

        Select the `region` in which the bucket was created.

        ### Reading

        When `watch` is set to `true` the S3Agent will watch the specified `bucket` for changes. An event will be emitted for every detected change.

        When `watch` is set to `false` the agent will emit an event for every file in the bucket on each sheduled run.

        #{emitting_file_handling_agent_description}

        ### Writing

        Specify the filename to use in `filename`, Liquid interpolation is possible to change the name per event.

        Use [Liquid](https://github.com/huginn/huginn/wiki/Formatting-Events-using-Liquid) templating in `data` to specify which part of the received event should be written.
      MD
    end

    event_description do
      "Events will looks like this:\n\n    %s" % if boolify(interpolated['watch'])
        Utils.pretty_print({
          "file_pointer" => {
            "file" => "filename",
            "agent_id" => id
          },
          "event_type" => "modified/added/removed"
        })
      else
        Utils.pretty_print({
          "file_pointer" => {
            "file" => "filename",
            "agent_id" => id
          }
        })
      end
    end

    def default_options
      {
        'mode' => 'read',
        'access_key_id' => '',
        'access_key_secret' => '',
        'watch' => 'true',
        'bucket' => "",
        'data' => '{{ data }}'
      }
    end

    form_configurable :mode, type: :array, values: %w(read write)
    form_configurable :access_key_id, roles: :validatable
    form_configurable :access_key_secret, roles: :validatable
    form_configurable :region, type: :array, values: %w(us-east-1 us-west-1 us-west-2 eu-west-1 eu-central-1 ap-southeast-1 ap-southeast-2 ap-northeast-1 ap-northeast-2 sa-east-1)
    form_configurable :watch, type: :array, values: %w(true false)
    form_configurable :bucket, roles: :completable
    form_configurable :filename
    form_configurable :data

    def validate_options
      if options['mode'].blank? || !['read', 'write'].include?(options['mode'])
        errors.add(:base, "The 'mode' option is required and must be set to 'read' or 'write'")
      end
      if options['bucket'].blank?
        errors.add(:base, "The 'bucket' option is required.")
      end
      if options['region'].blank?
        errors.add(:base, "The 'region' option is required.")
      end

      case interpolated['mode']
      when 'read'
        if options['watch'].blank? || ![true, false].include?(boolify(options['watch']))
          errors.add(:base, "The 'watch' option is required and must be set to 'true' or 'false'")
        end
      when 'write'
        if options['filename'].blank?
          errors.add(:base, "filename must be specified in 'write' mode")
        end
        if options['data'].blank?
          errors.add(:base, "data must be specified in 'write' mode")
        end
      end
    end

    def validate_access_key_id
      !!buckets
    end

    def validate_access_key_secret
      !!buckets
    end

    def complete_bucket
      (buckets || []).collect { |room| {text: room.name, id: room.name} }
    end

    def working?
      checked_without_error?
    end

    def check
      return if interpolated['mode'] != 'read'
      contents = safely do
                   get_bucket_contents
                 end
      if boolify(interpolated['watch'])
        watch(contents)
      else
        contents.each do |key, _|
          create_event payload: get_file_pointer(key)
        end
      end
    end

    def get_io(file)
      client.get_object(bucket: interpolated['bucket'], key: file).body
    end

    def receive(incoming_events)
      return if interpolated['mode'] != 'write'
      incoming_events.each do |event|
        safely do
          mo = interpolated(event)
          client.put_object(bucket: mo['bucket'], key: mo['filename'], body: mo['data'])
        end
      end
    end

    private

    def safely
      yield
    rescue Aws::S3::Errors::AccessDenied => e
      error("Could not access '#{interpolated['bucket']}' #{e.class} #{e.message}")
    rescue Aws::S3::Errors::ServiceError =>e
      error("#{e.class}: #{e.message}")
    end

    def watch(contents)
      if last_check_at.nil?
        self.memory['seen_contents'] = contents
        return
      end

      new_memory = contents.dup

      memory['seen_contents'].each do |key, etag|
        if contents[key].blank?
          create_event payload: get_file_pointer(key).merge(event_type: :removed)
        elsif contents[key] != etag
          create_event payload: get_file_pointer(key).merge(event_type: :modified)
        end
        contents.delete(key)
      end
      contents.each do |key, etag|
        create_event payload: get_file_pointer(key).merge(event_type: :added)
      end

      self.memory['seen_contents'] = new_memory
    end

    def get_bucket_contents
      contents = {}
      client.list_objects(bucket: interpolated['bucket']).each do |response|
        response.contents.each do |file|
          contents[file.key] = file.etag
        end
      end
      contents
    end

    def client
      @client ||= Aws::S3::Client.new(credentials: Aws::Credentials.new(interpolated['access_key_id'], interpolated['access_key_secret']),
                                      region: interpolated['region'])
    end

    def buckets(log = false)
      @buckets ||= client.list_buckets.buckets
    rescue Aws::S3::Errors::ServiceError => e
      false
    end
  end
end