require 'utils' # Agent is the core class in Huginn, representing a configurable, schedulable, reactive system with memory that can # be sub-classed for many different purposes. Agents can emit Events, as well as receive them and react in many different ways. # The basic Agent API is detailed on the Huginn wiki: https://github.com/huginn/huginn/wiki/Creating-a-new-agent class Agent < ActiveRecord::Base include AssignableTypes include MarkdownClassAttributes include JsonSerializedField include RdbmsFunctions include WorkingHelpers include LiquidInterpolatable include HasGuid include DryRunnable include SortableEvents markdown_class_attributes :description, :event_description load_types_in "Agents" SCHEDULES = %w[ every_1m every_2m every_5m every_10m every_30m every_1h every_2h every_5h every_12h every_1d every_2d every_7d midnight 1am 2am 3am 4am 5am 6am 7am 8am 9am 10am 11am noon 1pm 2pm 3pm 4pm 5pm 6pm 7pm 8pm 9pm 10pm 11pm never ] EVENT_RETENTION_SCHEDULES = [ ["Forever", 0], ['1 hour', 1.hour], ['6 hours', 6.hours], ["1 day", 1.day], *( [2, 3, 4, 5, 7, 14, 21, 30, 45, 90, 180, 365].map { |n| ["#{n} days", n.days] } ) ] json_serialize :options, :memory validates_presence_of :name, :user validates_inclusion_of :keep_events_for, in: EVENT_RETENTION_SCHEDULES.map(&:last) validates :sources, owned_by: :user_id validates :receivers, owned_by: :user_id validates :controllers, owned_by: :user_id validates :control_targets, owned_by: :user_id validates :scenarios, owned_by: :user_id validate :validate_schedule validate :validate_options after_initialize :set_default_schedule before_validation :set_default_schedule before_validation :unschedule_if_cannot_schedule before_save :unschedule_if_cannot_schedule before_create :set_last_checked_event_id after_save :possibly_update_event_expirations belongs_to :user, inverse_of: :agents belongs_to :service, inverse_of: :agents, optional: true has_many :events, -> { order("events.id desc") }, dependent: :delete_all, inverse_of: :agent has_one :most_recent_event, -> { order("events.id desc") }, inverse_of: :agent, class_name: "Event" has_many :logs, -> { order("agent_logs.id desc") }, dependent: :delete_all, inverse_of: :agent, class_name: "AgentLog" has_many :links_as_source, dependent: :delete_all, foreign_key: "source_id", class_name: "Link", inverse_of: :source has_many :links_as_receiver, dependent: :delete_all, foreign_key: "receiver_id", class_name: "Link", inverse_of: :receiver has_many :sources, through: :links_as_receiver, class_name: "Agent", inverse_of: :receivers has_many :received_events, -> { order("events.id desc") }, through: :sources, class_name: "Event", source: :events has_many :receivers, through: :links_as_source, class_name: "Agent", inverse_of: :sources has_many :control_links_as_controller, dependent: :delete_all, foreign_key: 'controller_id', class_name: 'ControlLink', inverse_of: :controller has_many :control_links_as_control_target, dependent: :delete_all, foreign_key: 'control_target_id', class_name: 'ControlLink', inverse_of: :control_target has_many :controllers, through: :control_links_as_control_target, class_name: "Agent", inverse_of: :control_targets has_many :control_targets, through: :control_links_as_controller, class_name: "Agent", inverse_of: :controllers has_many :scenario_memberships, dependent: :destroy, inverse_of: :agent has_many :scenarios, through: :scenario_memberships, inverse_of: :agents scope :active, -> { where(disabled: false, deactivated: false) } scope :inactive, -> { where(disabled: true).or(where(deactivated: true)) } scope :of_type, ->(type) { case type when Agent where(type: type.class.to_s) else where(type: type.to_s) end } def short_type type.demodulize end def check # Implement me in your subclass of Agent. end def default_options # Implement me in your subclass of Agent. {} end def receive(events) # Implement me in your subclass of Agent. end def is_form_configurable? false end def receive_web_request(params, method, format) # Implement me in your subclass of Agent. ["not implemented", 404, "text/plain", {}] # last two elements in response array are optional end # alternate method signature for receive_web_request # def receive_web_request(request=ActionDispatch::Request.new( ... )) # end # Implement me in your subclass to decide if your Agent is working. def working? raise "Implement me in your subclass" end def build_event(event) event = events.build(event) if event.is_a?(Hash) event.agent = self event.user = user event.expires_at ||= new_event_expiration_date event end def create_event(event) if can_create_events? event = build_event(event) event.save! event else error "This Agent cannot create events!" end end def credential(name) @credential_cache ||= {} if @credential_cache.has_key?(name) @credential_cache[name] else @credential_cache[name] = user.user_credentials.where(credential_name: name).first.try(:credential_value) end end def reload @credential_cache = {} super end def new_event_expiration_date keep_events_for > 0 ? keep_events_for.seconds.from_now : nil end def update_event_expirations! if keep_events_for == 0 events.update_all expires_at: nil else events.update_all "expires_at = " + rdbms_date_add("created_at", "SECOND", keep_events_for.to_i) end end def trigger_web_request(request) params = request.params.except(:action, :controller, :agent_id, :user_id, :format) if respond_to?(:receive_webhook) Rails.logger.warn "DEPRECATED: The .receive_webhook method is deprecated, please switch your Agent to use .receive_web_request." receive_webhook(params).tap do self.last_web_request_at = Time.now save! end else handled_request = if method(:receive_web_request).arity == 1 receive_web_request(request) else receive_web_request(params, request.method_symbol.to_s, request.format.to_s) end handled_request.tap do self.last_web_request_at = Time.now save! end end end def unavailable? disabled? || dependencies_missing? end def dependencies_missing? self.class.dependencies_missing? end def default_schedule self.class.default_schedule end def cannot_be_scheduled? self.class.cannot_be_scheduled? end def can_be_scheduled? !cannot_be_scheduled? end def cannot_receive_events? self.class.cannot_receive_events? end def can_receive_events? !cannot_receive_events? end def cannot_create_events? self.class.cannot_create_events? end def can_create_events? !cannot_create_events? end def can_control_other_agents? self.class.can_control_other_agents? end def can_dry_run? self.class.can_dry_run? end def no_bulk_receive? self.class.no_bulk_receive? end def log(message, options = {}) AgentLog.log_for_agent(self, message, options.merge(inbound_event: current_event)) end def error(message, options = {}) log(message, options.merge(level: 4)) end def delete_logs! logs.delete_all update_column :last_error_log_at, nil end def drop_pending_events false end def drop_pending_events=(bool) set_last_checked_event_id if bool end # Callbacks def set_default_schedule self.schedule = default_schedule unless schedule.present? || cannot_be_scheduled? end def unschedule_if_cannot_schedule self.schedule = nil if cannot_be_scheduled? end def set_last_checked_event_id if can_receive_events? && newest_event_id = Event.maximum(:id) self.last_checked_event_id = newest_event_id end end def possibly_update_event_expirations update_event_expirations! if saved_change_to_keep_events_for? end # Validation Methods private attr_accessor :current_event def validate_schedule unless cannot_be_scheduled? errors.add(:schedule, "is not a valid schedule") unless SCHEDULES.include?(schedule.to_s) end end def validate_options # Implement me in your subclass to test for valid options. end # Utility Methods def boolify(option_value) case option_value when true, 'true' true when false, 'false' false else nil end end def is_positive_integer?(value) Integer(value) >= 0 rescue StandardError false end # Class Methods class << self def build_clone(original) new(original.slice( :type, :options, :service_id, :schedule, :controller_ids, :control_target_ids, :source_ids, :receiver_ids, :keep_events_for, :propagate_immediately, :scenario_ids )) { |clone| # Give it a unique name 2.step do |i| name = '%s (%d)' % [original.name, i] unless exists?(name:) clone.name = name break end end } end def cannot_be_scheduled! @cannot_be_scheduled = true end def cannot_be_scheduled? !!@cannot_be_scheduled end def default_schedule(schedule = nil) @default_schedule = schedule unless schedule.nil? @default_schedule end def cannot_create_events! @cannot_create_events = true end def cannot_create_events? !!@cannot_create_events end def cannot_receive_events! @cannot_receive_events = true end def cannot_receive_events? !!@cannot_receive_events end def can_control_other_agents! @can_control_other_agents = true end def can_control_other_agents? !!@can_control_other_agents end def can_dry_run! @can_dry_run = true end def can_dry_run? !!@can_dry_run end def no_bulk_receive! @no_bulk_receive = true end def no_bulk_receive? !!@no_bulk_receive end def gem_dependency_check @gem_dependencies_checked = true @gem_dependencies_met = yield end def dependencies_missing? @gem_dependencies_checked && !@gem_dependencies_met end # Find all Agents that have received Events since the last execution of this method. Update those Agents with # their new `last_checked_event_id` and queue each of the Agents to be called with #receive using `async_receive`. # This is called by bin/schedule.rb periodically. def receive!(options = {}) Agent.transaction do scope = Agent .select("agents.id AS receiver_agent_id, sources.type AS source_agent_type, agents.type AS receiver_agent_type, events.id AS event_id") .joins("JOIN links ON (links.receiver_id = agents.id)") .joins("JOIN agents AS sources ON (links.source_id = sources.id)") .joins("JOIN events ON (events.agent_id = sources.id AND events.id > links.event_id_at_creation)") .where("NOT agents.disabled AND NOT agents.deactivated AND (agents.last_checked_event_id IS NULL OR events.id > agents.last_checked_event_id)") if options[:only_receivers].present? scope = scope.where("agents.id in (?)", options[:only_receivers]) end sql = scope.to_sql agents_to_events = {} Agent.connection.select_rows(sql).each do |receiver_agent_id, source_agent_type, receiver_agent_type, event_id| begin Object.const_get(source_agent_type) Object.const_get(receiver_agent_type) rescue NameError next end agents_to_events[receiver_agent_id.to_i] ||= [] agents_to_events[receiver_agent_id.to_i] << event_id end Agent.where(id: agents_to_events.keys).each do |agent| event_ids = agents_to_events[agent.id].uniq agent.update_attribute :last_checked_event_id, event_ids.max if agent.no_bulk_receive? event_ids.each { |event_id| Agent.async_receive(agent.id, [event_id]) } else Agent.async_receive(agent.id, event_ids) end end { agent_count: agents_to_events.keys.length, event_count: agents_to_events.values.flatten.uniq.compact.length } end end # This method will enqueue an AgentReceiveJob job. It accepts Agent and Event ids instead of a literal ActiveRecord # models because it is preferable to serialize jobs with ids. def async_receive(agent_id, event_ids) AgentReceiveJob.perform_later(agent_id, event_ids) end # Given a schedule name, run `check` via `bulk_check` on all Agents with that schedule. # This is called by bin/schedule.rb for each schedule in `SCHEDULES`. def run_schedule(schedule) return if schedule == 'never' types = where(schedule:).group(:type).pluck(:type) types.each do |type| next unless valid_type?(type) type.constantize.bulk_check(schedule) end end # Schedule `async_check`s for every Agent on the given schedule. This is normally called by `run_schedule` once # per type of agent, so you can override this to define custom bulk check behavior for your custom Agent type. def bulk_check(schedule) raise "Call #bulk_check on the appropriate subclass of Agent" if self == Agent where("NOT disabled AND NOT deactivated AND schedule = ?", schedule).pluck("agents.id").each do |agent_id| async_check(agent_id) end end # This method will enqueue an AgentCheckJob job. It accepts an Agent id instead of a literal Agent because it is # preferable to serialize job with ids, instead of with the full Agents. def async_check(agent_id) AgentCheckJob.perform_later(agent_id) end end public def to_liquid Drop.new(self) end class Drop < LiquidDroppable::Drop def type @object.short_type end METHODS = %i[ id name type options memory sources receivers schedule controllers control_targets disabled keep_events_for propagate_immediately ] METHODS.each { |attr| define_method(attr) { @object.__send__(attr) } unless method_defined?(attr) } def working @object.working? end def url Rails.application.routes.url_helpers.agent_url( @object, Rails.application.config.action_mailer.default_url_options ) end end end