diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/services/process_feed_service.rb | 393 |
1 files changed, 205 insertions, 188 deletions
diff --git a/app/services/process_feed_service.rb b/app/services/process_feed_service.rb index c335d2159..19ab0c858 100644 --- a/app/services/process_feed_service.rb +++ b/app/services/process_feed_service.rb @@ -22,268 +22,285 @@ class ProcessFeedService < BaseService class ProcessEntry def call(xml, account) @account = account - @xml = xml - - return if skip_unsupported_type? - - case verb - when :post, :share - return create_status - when :delete - return delete_status - end + @fetched = Activity.new(xml) + + return unless [:activity, :note, :comment].include?(@fetched.type) + + klass = case @fetched.verb + when :post + PostActivity + when :share + ShareActivity + when :delete + DeletionActivity + 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 - private - - def create_status - if redis.exists("delete_upon_arrival:#{@account.id}:#{id}") - Rails.logger.debug "Delete for status #{id} was queued, ignoring" - return + class Activity + def initialize(xml, account = nil) + @xml = xml + @account = account end - status, just_created = nil + def verb + raw = @xml.at_xpath('./activity:verb', activity: TagManager::AS_XMLNS).content + TagManager::VERBS.key(raw) + rescue + :post + end - Rails.logger.debug "Creating remote status #{id}" + def type + raw = @xml.at_xpath('./activity:object-type', activity: TagManager::AS_XMLNS).content + TagManager::TYPES.key(raw) + rescue + :activity + end - if verb == :share - original_status = shared_status_from_xml(@xml.at_xpath('.//activity:object', activity: TagManager::AS_XMLNS)) - return nil if original_status.nil? + def id + @xml.at_xpath('./xmlns:id', xmlns: TagManager::XMLNS).content end - ApplicationRecord.transaction do - status, just_created = status_from_xml(@xml) + def url + link = @xml.at_xpath('./xmlns:link[@rel="alternate"]', xmlns: TagManager::XMLNS) + link.nil? ? nil : link['href'] + end - return if status.nil? - return status unless just_created + private - if verb == :share - status.reblog = original_status.reblog? ? original_status.reblog : original_status + def find_status(uri) + if TagManager.instance.local_id?(uri) + local_id = TagManager.instance.unique_tag_to_local_id(uri, 'Status') + return Status.find_by(id: local_id) end - status.save! + Status.find_by(uri: uri) 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) + def redis + Redis.current end + end - notify_about_mentions!(status) unless status.reblog? - notify_about_reblog!(status) if status.reblog? && status.reblog.account.local? + 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 "Queuing remote status #{status.id} (#{id}) for distribution" + 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 - LinkCrawlWorker.perform_async(status.id) unless status.spoiler_text? - DistributionWorker.perform_async(status.id) + Rails.logger.debug "Queuing remote status #{status.id} (#{id}) for distribution" - status - end + LinkCrawlWorker.perform_async(status.id) unless status.spoiler_text? + DistributionWorker.perform_async(status.id) - 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) + [status, true] end - end - def notify_about_reblog!(status) - NotifyService.new.call(status.reblog.account, status) - end + def content + @xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS).content + end - def delete_status - Rails.logger.debug "Deleting remote status #{id}" - status = Status.find_by(uri: id, account: @account) + def content_language + @xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS)['xml:lang']&.presence || 'en' + end - if status.nil? - redis.setex("delete_upon_arrival:#{@account.id}:#{id}", 6 * 3_600, id) - else - RemoveStatusService.new.call(status) + def content_warning + @xml.at_xpath('./xmlns:summary', xmlns: TagManager::XMLNS)&.content || '' end - end - def skip_unsupported_type? - !([:post, :share, :delete].include?(verb) && [:activity, :note, :comment].include?(type)) - end + def visibility_scope + @xml.at_xpath('./mastodon:scope', mastodon: TagManager::MTDN_XMLNS)&.content&.to_sym || :public + end - def shared_status_from_xml(entry) - status = find_status(id(entry)) + def published + @xml.at_xpath('./xmlns:published', xmlns: TagManager::XMLNS).content + end - return status unless status.nil? + def thread? + !@xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS).nil? + end - FetchRemoteStatusService.new.call(url(entry)) - end + def thread + thr = @xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS) + [thr['ref'], thr['href']] + end - def status_from_xml(entry) - # Return early if status already exists in db - status = find_status(id(entry)) + private - return [status, false] unless status.nil? + def converstation_to_persistent + uri = @xml.at_xpath('./ostatus:conversation', ostatus: TagManager::OS_XMLNS)&.attribute('ref')&.content + return if uri.nil? - account = @account + 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 - return [nil, false] if account.suspended? + found = Conversation.find_by(uri: uri) + found ? [found, false] : [Conversation.create!(uri: uri), true] + end - 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 - ) + def save_mentions(parent) + processed_account_ids = [] - mentions_from_xml(status, entry) - hashtags_from_xml(status, entry) - media_from_xml(status, entry) + @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'] - [status, true] - end + mentioned_account = account_from_href(link['href']) - def find_or_create_conversation(xml) - uri = xml.at_xpath('./ostatus:conversation', ostatus: TagManager::OS_XMLNS)&.attribute('ref')&.content - return if uri.nil? + next if mentioned_account.nil? || processed_account_ids.include?(mentioned_account.id) - 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 + mentioned_account.mentions.where(status: parent).first_or_create(status: parent) - Conversation.find_by(uri: uri) || Conversation.create!(uri: uri) - end + # So we can skip duplicate mentions + processed_account_ids << mentioned_account.id + end + end - def find_status(uri) - if TagManager.instance.local_id?(uri) - local_id = TagManager.instance.unique_tag_to_local_id(uri, 'Status') - return Status.find_by(id: local_id) + def save_hashtags(parent) + tags = @xml.xpath('./xmlns:category', xmlns: TagManager::XMLNS).map { |category| category['term'] }.select(&:present?) + ProcessHashtagsService.new.call(parent, tags) end - Status.find_by(uri: uri) - end + def save_media(parent) + do_not_download = DomainBlock.find_by(domain: parent.account.domain)&.reject_media? - def mentions_from_xml(parent, xml) - processed_account_ids = [] + @xml.xpath('./xmlns:link[@rel="enclosure"]', xmlns: TagManager::XMLNS).each do |link| + next unless link['href'] - 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'] + media = MediaAttachment.where(status: parent, remote_url: link['href']).first_or_initialize(account: parent.account, status: parent, remote_url: link['href']) + parsed_url = Addressable::URI.parse(link['href']).normalize - mentioned_account = account_from_href(link['href']) + next if !%w(http https).include?(parsed_url.scheme) || parsed_url.host.empty? - next if mentioned_account.nil? || processed_account_ids.include?(mentioned_account.id) + media.save - mentioned_account.mentions.where(status: parent).first_or_create(status: parent) + next if do_not_download - # So we can skip duplicate mentions - processed_account_ids << mentioned_account.id + begin + media.file_remote_url = link['href'] + media.save! + rescue ActiveRecord::RecordInvalid + next + end + end end - end - def account_from_href(href) - url = Addressable::URI.parse(href).normalize + def account_from_href(href) + 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) + 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 - 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 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| - 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']) - parsed_url = Addressable::URI.parse(link['href']).normalize - - next if !%w(http https).include?(parsed_url.scheme) || parsed_url.host.empty? - - media.save - - next if do_not_download - - begin - media.file_remote_url = link['href'] - media.save! - rescue ActiveRecord::RecordInvalid - next - end + class ShareActivity < CreationActivity + def perform + status, just_created = super + NotifyService.new.call(reblog.account, status) if reblog&.account&.local? && just_created + status end - end - def id(xml = @xml) - xml.at_xpath('./xmlns:id', xmlns: TagManager::XMLNS).content - end + def object + @xml.at_xpath('.//activity:object', activity: TagManager::AS_XMLNS) + end - def verb(xml = @xml) - raw = xml.at_xpath('./activity:verb', activity: TagManager::AS_XMLNS).content - TagManager::VERBS.key(raw) - rescue - :post - end + private - def type(xml = @xml) - raw = xml.at_xpath('./activity:object-type', activity: TagManager::AS_XMLNS).content - TagManager::TYPES.key(raw) - rescue - :activity - end + def status + reblog && super + end - def url(xml = @xml) - link = xml.at_xpath('./xmlns:link[@rel="alternate"]', xmlns: TagManager::XMLNS) - link.nil? ? nil : link['href'] - end + def reblog + return @reblog if defined? @reblog - def content(xml = @xml) - 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 - def content_language(xml = @xml) - xml.at_xpath('./xmlns:content', xmlns: TagManager::XMLNS)['xml:lang']&.presence || 'en' - end + class PostActivity < CreationActivity + def perform + status, just_created = super - def content_warning(xml = @xml) - xml.at_xpath('./xmlns:summary', xmlns: TagManager::XMLNS)&.content || '' - end + 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 - def visibility_scope(xml = @xml) - xml.at_xpath('./mastodon:scope', mastodon: TagManager::MTDN_XMLNS)&.content&.to_sym || :public - end + status + end - def published(xml = @xml) - xml.at_xpath('./xmlns:published', xmlns: TagManager::XMLNS).content - end + private - def thread?(xml = @xml) - !xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS).nil? + def reblog + nil + end end - def thread(xml = @xml) - thr = xml.at_xpath('./thr:in-reply-to', thr: TagManager::THR_XMLNS) - [thr['ref'], thr['href']] - end + class DeletionActivity < Activity + def perform + Rails.logger.debug "Deleting remote status #{id}" + status = Status.find_by(uri: id, account: @account) - def account?(xml = @xml) - !xml.at_xpath('./xmlns:author', xmlns: TagManager::XMLNS).nil? + if status.nil? + redis.setex("delete_upon_arrival:#{@account.id}:#{id}", 6 * 3_600, id) + else + RemoveStatusService.new.call(status) + end + end end - def redis - Redis.current + class RemoteActivity < Activity + def perform + find_status(id) || FetchRemoteStatusService.new.call(url) + end end end end |