# 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.

# Ruby SDK for convenient access of PredictionIO Output API.
#
# Author::    PredictionIO Team (support@prediction.io)
# Copyright:: Copyright (c) 2014 TappingStone, Inc.
# License::   Apache License, Version 2.0

require 'date'
require "net/http"
require 'active_support/core_ext/hash'

module PredictionIO
  # This class contains methods that interface with the PredictionIO Event
  # Server via the PredictionIO Event API using REST requests.
  #
  # Many REST request methods support optional arguments. They can be supplied
  # to these methods as Hash'es. For a complete reference, please visit
  # http://prediction.io.
  #
  # == High-performance Asynchronous Backend
  #
  # All REST request methods come in both synchronous and asynchronous flavors.
  # Both flavors accept the same set of arguments. In addition, all synchronous
  # request methods can instead accept a PredictionIO::AsyncResponse object
  # generated from asynchronous request methods as its first argument. In this
  # case, the method will block until a response is received from it.
  #
  # Any network reconnection and request retry is automatically handled in the
  # background. Exceptions will be thrown after a request times out to avoid
  # infinite blocking.
  #
  # == Installation
  # The easiest way is to use RubyGems:
  #     gem install predictionio
  #
  # == Synopsis
  # In most cases, using synchronous methods. If you have a special performance
  # requirement, you may want to take a look at asynchronous methods.
  #
  # === Instantiate an EventClient
  #     # Include the PredictionIO SDK
  #     require 'predictionio'
  #
  #     client = PredictionIO::EventClient.new(<access_key>)
  #
  # === Import a User Record from Your App (with asynchronous/non-blocking
  #     requests)
  #
  #     #
  #     # (your user registration logic)
  #     #
  #
  #     uid = get_user_from_your_db()
  #
  #     # PredictionIO call to create user
  #     response = client.aset_user(uid)
  #
  #     #
  #     # (other work to do for the rest of the page)
  #     #
  #
  #     begin
  #       # PredictionIO call to retrieve results from an asynchronous response
  #       result = client.set_user(response)
  #     rescue PredictionIO::EventClient::NotCreatedError => e
  #       log_and_email_error(...)
  #     end
  #
  # === Import a User Action (Rate) from Your App (with synchronous/blocking
  #     requests)
  #     # PredictionIO call to record the view action
  #     begin
  #       result = client.record_user_action_on_item('rate', 'foouser',
  #                                                  'baritem',
  #                                                  'rating' => 4)
  #     rescue PredictionIO::EventClient::NotCreatedError => e
  #       ...
  #     end
  class EventClient
    # Raised when an event is not created after a synchronous API call.
    class NotCreatedError < StandardError; end

    # Create a new PredictionIO Event Client with defaults:
    # - 1 concurrent HTTP(S) connections (threads)
    # - API entry point at http://localhost:7070 (apiurl)
    # - a 60-second timeout for each HTTP(S) connection (thread_timeout)
    def initialize(access_key, apiurl = 'http://localhost:7070', threads = 1, thread_timeout = 60)
      @access_key = access_key
      @http = PredictionIO::Connection.new(URI(apiurl), threads, thread_timeout)
    end

    # Returns the number of pending requests within the current client.
    def pending_requests
      @http.packages.size
    end

    # Returns PredictionIO's status in string.
    def get_status
      status = @http.aget(PredictionIO::AsyncRequest.new('/')).get
      begin
        status.body
      rescue
        status
      end
    end

    # :category: Asynchronous Methods
    # Asynchronously request to create an event and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #create_event.
    def acreate_event(event, entity_type, entity_id, optional = {})
      h = optional.with_indifferent_access
      h.key?('eventTime') || h['eventTime'] = DateTime.now.iso8601
      h['event'] = event
      h['entityType'] = entity_type
      h['entityId'] = entity_id
      @http.apost(PredictionIO::AsyncRequest.new(event_path_with_main_params(h), h.except(:channel).to_json))
    end

    def event_path_with_main_params(options)
      result = "/events.json?accessKey=#{@access_key}"
      result += "&channel=#{options[:channel]}" if options[:channel]

      result
    end

    # :category: Asynchronous Methods
    # Asynchronously request to create a batch of event and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /batch/events.json
    #
    # See also #acreate_event.
    def acreate_batch_event(payload)
      @http.apost(PredictionIO::AsyncRequest.new(
        "/batch/events.json?accessKey=#{@access_key}", payload.to_json
      ))
    end

    # :category: Synchronous Methods
    # Synchronously request to create an event and block until a response is
    # received.
    #
    # See also #acreate_event.
    #
    # call-seq:
    # create_event(event, entity_type, entity_id, optional = {})
    # create_event(async_response)
    def create_event(*args)
      sync_events(:acreate_event, *args)
    end

    # :category: Synchronous Methods
    # Synchronously request to create an event and block until a response is
    # received.
    #
    # See also #acreate_event.
    #
    # call-seq:
    # create_batch_event(payload)
    # create_batch_event(async_response)
    def create_batch_event(payload)
      sync_events(:acreate_batch_event, payload)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to set properties of a user and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #set_user.
    def aset_user(uid, optional = {})
      acreate_event('$set', 'user', uid, optional)
    end

    # :category: Synchronous Methods
    # Synchronously request to set properties of a user and block until a
    # response is received.
    #
    # See also #aset_user.
    #
    # call-seq:
    # set_user(uid, optional = {})
    # set_user(async_response)
    def set_user(*args)
      sync_events(:aset_user, *args)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to unset properties of a user and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # properties must be a non-empty Hash.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #unset_user.
    def aunset_user(uid, optional)
      optional.key?('properties') ||
        fail(ArgumentError, 'properties must be present when event is $unset')
      optional['properties'].empty? &&
        fail(ArgumentError, 'properties cannot be empty when event is $unset')
      acreate_event('$unset', 'user', uid, optional)
    end

    # :category: Synchronous Methods
    # Synchronously request to unset properties of a user and block until a
    # response is received.
    #
    # See also #aunset_user.
    #
    # call-seq:
    # unset_user(uid, optional)
    # unset_user(async_response)
    def unset_user(*args)
      sync_events(:aunset_user, *args)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to delete a user and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #delete_user.
    def adelete_user(uid)
      acreate_event('$delete', 'user', uid)
    end

    # :category: Synchronous Methods
    # Synchronously request to delete a user and block until a response is
    # received.
    #
    # See also #adelete_user.
    #
    # call-seq:
    # delete_user(uid)
    # delete_user(async_response)
    def delete_user(*args)
      sync_events(:adelete_user, *args)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to set properties of an item and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #set_item.
    def aset_item(iid, optional = {})
      acreate_event('$set', 'item', iid, optional)
    end

    # :category: Synchronous Methods
    # Synchronously request to set properties of an item and block until a
    # response is received.
    #
    # See also #aset_item.
    #
    # call-seq:
    # set_item(iid, properties = {}, optional = {})
    # set_item(async_response)
    def set_item(*args)
      sync_events(:aset_item, *args)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to unset properties of an item and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # properties must be a non-empty Hash.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #unset_item.
    def aunset_item(iid, optional)
      optional.key?('properties') ||
        fail(ArgumentError, 'properties must be present when event is $unset')
      optional['properties'].empty? &&
        fail(ArgumentError, 'properties cannot be empty when event is $unset')
      acreate_event('$unset', 'item', iid, optional)
    end

    # :category: Synchronous Methods
    # Synchronously request to unset properties of an item and block until a
    # response is received.
    #
    # See also #aunset_item.
    #
    # call-seq:
    # unset_item(iid, properties, optional = {})
    # unset_item(async_response)
    def unset_item(*args)
      sync_events(:aunset_item, *args)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to delete an item and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #delete_item.
    def adelete_item(uid)
      acreate_event('$delete', 'item', uid)
    end

    # :category: Synchronous Methods
    # Synchronously request to delete an item and block until a response is
    # received.
    #
    # See also #adelete_item.
    #
    # call-seq:
    # delete_item(uid)
    # delete_item(async_response)
    def delete_item(*args)
      sync_events(:adelete_item, *args)
    end

    # :category: Asynchronous Methods
    # Asynchronously request to record an action on an item and return a
    # PredictionIO::AsyncResponse object immediately.
    #
    # Corresponding REST API method: POST /events.json
    #
    # See also #record_user_action_on_item.
    def arecord_user_action_on_item(action, uid, iid, optional = {})
      optional['targetEntityType'] = 'item'
      optional['targetEntityId'] = iid
      acreate_event(action, 'user', uid, optional)
    end

    # :category: Synchronous Methods
    # Synchronously request to record an action on an item and block until a
    # response is received.
    #
    # See also #arecord_user_action_on_item.
    #
    # call-seq:
    # record_user_action_on_item(action, uid, iid, optional = {})
    # record_user_action_on_item(async_response)
    def record_user_action_on_item(*args)
      sync_events(:arecord_user_action_on_item, *args)
    end

    protected

    # Internal helper method. Do not call directly.
    def sync_events(sync_m, *args)
      if args[0].is_a?(PredictionIO::AsyncResponse)
        response = args[0].get
      else
        response = send(sync_m, *args).get
      end
      return response if response.is_a?(Net::HTTPCreated)
      begin
        msg = response.body
      rescue
        raise NotCreatedError, response
      end
      fail NotCreatedError, msg
    end
  end
end