Revert "Wrap methods of ProcessFeedService::ProcessEntry in classes (#4151)" (#4157)

This reverts commit 425acecfdb.

app/services/process_feed_service.rb

@@ -22,59 +22,145 @@ class ProcessFeedService < BaseService
   class ProcessEntry
     def call(xml, account)
       @account = account
-      @fetched = Activity.new(xml)
+      @xml     = xml
 
-      return unless [:activity, :note, :comment].include?(@fetched.type)
+      return if skip_unsupported_type?
 
-      klass = case @fetched.verb
+      case verb
-              when :post
+      when :post, :share
-                PostActivity
+        return create_status
-              when :share
-                ShareActivity
       when :delete
-                DeletionActivity
+        return delete_status
-              else
-                return
       end
-
-      @fetched = klass.new(xml, account)
-      @fetched.perform
     rescue ActiveRecord::RecordInvalid => e
       Rails.logger.debug "Nothing was saved for #{id} because: #{e}"
       nil
     end
 
-    class Activity
-      def initialize(xml, account = nil)
-        @xml = xml
-        @account = account
-      end
-
-      def verb
-        raw = @xml.at_xpath('./activity:verb', activity: TagManager::AS_XMLNS).content
-        TagManager::VERBS.key(raw)
-      rescue
-        :post
-      end
-
-      def type
-        raw = @xml.at_xpath('./activity:object-type', activity: TagManager::AS_XMLNS).content
-        TagManager::TYPES.key(raw)
-      rescue
-        :activity
-      end
-
-      def id
-        @xml.at_xpath('./xmlns:id', xmlns: TagManager::XMLNS).content
-      end
-
-      def url
-        link = @xml.at_xpath('./xmlns:link[@rel="alternate"]', xmlns: TagManager::XMLNS)
-        link.nil? ? nil : link['href']
-      end
-
     private
 
+    def create_status
+      if redis.exists("delete_upon_arrival:#{@account.id}:#{id}")
+        Rails.logger.debug "Delete for status #{id} was queued, ignoring"
+        return
+      end
+
+      status, just_created = nil
+
+      Rails.logger.debug "Creating remote status #{id}"
+
+      if verb == :share
+        original_status = shared_status_from_xml(@xml.at_xpath('.//activity:object', activity: TagManager::AS_XMLNS))
+        return nil if original_status.nil?
+      end
+
+      ApplicationRecord.transaction do
+        status, just_created = status_from_xml(@xml)
+
+        return if status.nil?
+        return status unless just_created
+
+        if verb == :share
+          status.reblog = original_status.reblog? ? original_status.reblog : original_status
+        end
+
+        status.save!
+      end
+
+      if thread?(@xml) && status.thread.nil?
+        Rails.logger.debug "Trying to attach #{status.id} (#{id(@xml)}) to #{thread(@xml).first}"
+        ThreadResolveWorker.perform_async(status.id, thread(@xml).second)
+      end
+
+      notify_about_mentions!(status) unless status.reblog?
+      notify_about_reblog!(status) if status.reblog? && status.reblog.account.local?
+
+      Rails.logger.debug "Queuing remote status #{status.id} (#{id}) for distribution"
+
+      LinkCrawlWorker.perform_async(status.id) unless status.spoiler_text?
+      DistributionWorker.perform_async(status.id)
+
+      status
+    end
+
+    def notify_about_mentions!(status)
+      status.mentions.includes(:account).each do |mention|
+        mentioned_account = mention.account
+        next unless mentioned_account.local?
+        NotifyService.new.call(mentioned_account, mention)
+      end
+    end
+
+    def notify_about_reblog!(status)
+      NotifyService.new.call(status.reblog.account, status)
+    end
+
+    def delete_status
+      Rails.logger.debug "Deleting remote status #{id}"
+      status = Status.find_by(uri: id, account: @account)
+
+      if status.nil?
+        redis.setex("delete_upon_arrival:#{@account.id}:#{id}", 6 * 3_600, id)
+      else
+        RemoveStatusService.new.call(status)
+      end
+    end
+
+    def skip_unsupported_type?
+      !([:post, :share, :delete].include?(verb) && [:activity, :note, :comment].include?(type))
+    end
+
+    def shared_status_from_xml(entry)
+      status = find_status(id(entry))
+
+      return status unless status.nil?
+
+      FetchRemoteStatusService.new.call(url(entry))
+    end
+
+    def status_from_xml(entry)
+      # Return early if status already exists in db
+      status = find_status(id(entry))
+
+      return [status, false] unless status.nil?
+
+      account = @account
+
+      return [nil, false] if account.suspended?
+
+      status = Status.create!(
+        uri: id(entry),
+        url: url(entry),
+        account: account,
+        text: content(entry),
+        spoiler_text: content_warning(entry),
+        created_at: published(entry),
+        reply: thread?(entry),
+        language: content_language(entry),
+        visibility: visibility_scope(entry),
+        conversation: find_or_create_conversation(entry),
+        thread: thread?(entry) ? find_status(thread(entry).first) : nil
+      )
+
+      mentions_from_xml(status, entry)
+      hashtags_from_xml(status, entry)
+      media_from_xml(status, entry)
+
+      [status, true]
+    end
+
+    def find_or_create_conversation(xml)
+      uri = xml.at_xpath('./ostatus:conversation', ostatus: TagManager::OS_XMLNS)&.attribute('ref')&.content
+      return if uri.nil?
+
+      if TagManager.instance.local_id?(uri)
+        local_id = TagManager.instance.unique_tag_to_local_id(uri, 'Conversation')
+        return Conversation.find_by(id: local_id)
+      end
+
+      Conversation.find_by(uri: uri) || Conversation.create!(uri: uri)
+    end
+
     def find_status(uri)
       if TagManager.instance.local_id?(uri)
         local_id = TagManager.instance.unique_tag_to_local_id(uri, 'Status')
@@ -84,106 +170,10 @@ class ProcessFeedService < BaseService
       Status.find_by(uri: uri)
     end
 
-      def redis
+    def mentions_from_xml(parent, xml)
-        Redis.current
-      end
-    end
-
-    class CreationActivity < Activity
-      def perform
-        if redis.exists("delete_upon_arrival:#{@account.id}:#{id}")
-          Rails.logger.debug "Delete for status #{id} was queued, ignoring"
-          return
-        end
-
-        Rails.logger.debug "Creating remote status #{id}"
-        # Return early if status already exists in db
-        status = find_status(id)
-
-        return [status, false] unless status.nil?
-
-        return [nil, false] if @account.suspended?
-
-        status = Status.create!(
-          uri: id,
-          url: url,
-          account: @account,
-          reblog: reblog,
-          text: content,
-          spoiler_text: content_warning,
-          created_at: published,
-          reply: thread?,
-          language: content_language,
-          visibility: visibility_scope,
-          conversation: converstation_to_persistent&.first,
-          thread: thread? ? find_status(thread.first) : nil
-        )
-
-        save_mentions(status)
-        save_hashtags(status)
-        save_media(status)
-
-        if thread? && status.thread.nil?
-          Rails.logger.debug "Trying to attach #{status.id} (#{id}) to #{thread.first}"
-          ThreadResolveWorker.perform_async(status.id, thread.second)
-        end
-
-        Rails.logger.debug "Queuing remote status #{status.id} (#{id}) for distribution"
-
-        LinkCrawlWorker.perform_async(status.id) unless status.spoiler_text?
-        DistributionWorker.perform_async(status.id)
-
-        [status, true]
-      end
-
-      def content
-        @xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS).content
-      end
-
-      def content_language
-        @xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS)['xml:lang']&.presence || 'en'
-      end
-
-      def content_warning
-        @xml.at_xpath('./xmlns:summary', xmlns: TagManager::XMLNS)&.content || ''
-      end
-
-      def visibility_scope
-        @xml.at_xpath('./mastodon:scope', mastodon: TagManager::MTDN_XMLNS)&.content&.to_sym || :public
-      end
-
-      def published
-        @xml.at_xpath('./xmlns:published', xmlns: TagManager::XMLNS).content
-      end
-
-      def thread?
-        !@xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS).nil?
-      end
-
-      def thread
-        thr = @xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS)
-        [thr['ref'], thr['href']]
-      end
-
-      private
-
-      def converstation_to_persistent
-        uri = @xml.at_xpath('./ostatus:conversation', ostatus: TagManager::OS_XMLNS)&.attribute('ref')&.content
-        return if uri.nil?
-
-        if TagManager.instance.local_id?(uri)
-          local_id = TagManager.instance.unique_tag_to_local_id(uri, 'Conversation')
-          return Conversation.find_by(id: local_id)
-        end
-
-        found = Conversation.find_by(uri: uri)
-        found ? [found, false] : [Conversation.create!(uri: uri), true]
-      end
-
-      def save_mentions(parent)
       processed_account_ids = []
 
-        @xml.xpath('./xmlns:link[@rel="mentioned"]', xmlns: TagManager::XMLNS).each do |link|
+      xml.xpath('./xmlns:link[@rel="mentioned"]', xmlns: TagManager::XMLNS).each do |link|
         next if [TagManager::TYPES[:group], TagManager::TYPES[:collection]].include? link['ostatus:object-type']
 
         mentioned_account = account_from_href(link['href'])
@@ -197,15 +187,25 @@ class ProcessFeedService < BaseService
       end
     end
 
-      def save_hashtags(parent)
+    def account_from_href(href)
-        tags = @xml.xpath('./xmlns:category', xmlns: TagManager::XMLNS).map { |category| category['term'] }.select(&:present?)
+      url = Addressable::URI.parse(href).normalize
+
+      if TagManager.instance.web_domain?(url.host)
+        Account.find_local(url.path.gsub('/users/', ''))
+      else
+        Account.where(uri: href).or(Account.where(url: href)).first || FetchRemoteAccountService.new.call(href)
+      end
+    end
+
+    def hashtags_from_xml(parent, xml)
+      tags = xml.xpath('./xmlns:category', xmlns: TagManager::XMLNS).map { |category| category['term'] }.select(&:present?)
       ProcessHashtagsService.new.call(parent, tags)
     end
 
-      def save_media(parent)
+    def media_from_xml(parent, xml)
       do_not_download = DomainBlock.find_by(domain: parent.account.domain)&.reject_media?
 
-        @xml.xpath('./xmlns:link[@rel="enclosure"]', xmlns: TagManager::XMLNS).each do |link|
+      xml.xpath('./xmlns:link[@rel="enclosure"]', xmlns: TagManager::XMLNS).each do |link|
         next unless link['href']
 
         media = MediaAttachment.where(status: parent, remote_url: link['href']).first_or_initialize(account: parent.account, status: parent, remote_url: link['href'])
@@ -226,81 +226,64 @@ class ProcessFeedService < BaseService
       end
     end
 
-      def account_from_href(href)
+    def id(xml = @xml)
-        url = Addressable::URI.parse(href).normalize
+      xml.at_xpath('./xmlns:id', xmlns: TagManager::XMLNS).content
-
-        if TagManager.instance.web_domain?(url.host)
-          Account.find_local(url.path.gsub('/users/', ''))
-        else
-          Account.where(uri: href).or(Account.where(url: href)).first || FetchRemoteAccountService.new.call(href)
-        end
-      end
     end
 
-    class ShareActivity < CreationActivity
+    def verb(xml = @xml)
-      def perform
+      raw = xml.at_xpath('./activity:verb', activity: TagManager::AS_XMLNS).content
-        status, just_created = super
+      TagManager::VERBS.key(raw)
-        NotifyService.new.call(reblog.account, status) if reblog&.account&.local? && just_created
+    rescue
-        status
+      :post
     end
 
-      def object
+    def type(xml = @xml)
-        @xml.at_xpath('.//activity:object', activity: TagManager::AS_XMLNS)
+      raw = xml.at_xpath('./activity:object-type', activity: TagManager::AS_XMLNS).content
+      TagManager::TYPES.key(raw)
+    rescue
+      :activity
     end
 
-      private
+    def url(xml = @xml)
-
+      link = xml.at_xpath('./xmlns:link[@rel="alternate"]', xmlns: TagManager::XMLNS)
-      def status
+      link.nil? ? nil : link['href']
-        reblog && super
     end
 
-      def reblog
+    def content(xml = @xml)
-        return @reblog if defined? @reblog
+      xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS).content
-
-        original_status = RemoteActivity.new(object).perform
-        @reblog = original_status.reblog? ? original_status.reblog : original_status
-      end
     end
 
-    class PostActivity < CreationActivity
+    def content_language(xml = @xml)
-      def perform
+      xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS)['xml:lang']&.presence || 'en'
-        status, just_created = super
-
-        if just_created
-          status.mentions.includes(:account).each do |mention|
-            mentioned_account = mention.account
-            next unless mentioned_account.local?
-            NotifyService.new.call(mentioned_account, mention)
-          end
     end
 
-        status
+    def content_warning(xml = @xml)
+      xml.at_xpath('./xmlns:summary', xmlns: TagManager::XMLNS)&.content || ''
     end
 
-      private
+    def visibility_scope(xml = @xml)
-
+      xml.at_xpath('./mastodon:scope', mastodon: TagManager::MTDN_XMLNS)&.content&.to_sym || :public
-      def reblog
-        nil
-      end
     end
 
-    class DeletionActivity < Activity
+    def published(xml = @xml)
-      def perform
+      xml.at_xpath('./xmlns:published', xmlns: TagManager::XMLNS).content
-        Rails.logger.debug "Deleting remote status #{id}"
-        status = Status.find_by(uri: id, account: @account)
-
-        if status.nil?
-          redis.setex("delete_upon_arrival:#{@account.id}:#{id}", 6 * 3_600, id)
-        else
-          RemoveStatusService.new.call(status)
-        end
-      end
     end
 
-    class RemoteActivity < Activity
+    def thread?(xml = @xml)
-      def perform
+      !xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS).nil?
-        find_status(id) || FetchRemoteStatusService.new.call(url)
     end
+
+    def thread(xml = @xml)
+      thr = xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS)
+      [thr['ref'], thr['href']]
+    end
+
+    def account?(xml = @xml)
+      !xml.at_xpath('./xmlns:author', xmlns: TagManager::XMLNS).nil?
+    end
+
+    def redis
+      Redis.current
     end
   end
 end