MetuFSS
/
metu.life
2
1
Fork 0
Code Issues Pull Requests Releases Wiki Activity

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

Browse Source 
ProcessFeedService::ProcessEntry had many methods, so wrap them in classes
representing activities.
master
Akihiko Odaki (@fn_aki@pawoo.net) 7 years ago committed by Eugen Rochko
parent 29f314a502
commit 425acecfdb
1 changed files with 205 additions and 188 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 393
      app/services/process_feed_service.rb

393
app/services/process_feed_service.rb
Unescape View File

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

Write Preview
Loading…
Cancel
Save