autonomic-cooperative/hometown
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge tag 'v3.0.0' into hometown-dev

Browse Source
This commit is contained in:
Darius Kazemi
2019-10-08 13:24:20 -07:00
parent bfd29a34eb 83d3e7733d
commit 228e0f0f6e
1012 changed files with 31176 additions and 15165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
app/services/account_search_service.rb
View File

@ -4,51 +4,158 @@ class AccountSearchService < BaseService
attr_reader :query, :limit, :offset, :options, :account
def call(query, account = nil, options = {})
@query = query.strip
@limit = options[:limit].to_i
@offset = options[:offset].to_i
@options = options
@account = account
@acct_hint = query.start_with?('@')
@query = query.strip.gsub(/\A@/, '')
@limit = options[:limit].to_i
@offset = options[:offset].to_i
@options = options
@account = account
search_service_results
search_service_results.compact.uniq
end
private
def search_service_results
return [] if query_blank_or_hashtag? || limit < 1
return [] if query.blank? || limit < 1
if resolving_non_matching_remote_account?
[ResolveAccountService.new.call("#{query_username}@#{query_domain}")].compact
else
search_results_and_exact_match.compact.uniq.slice(0, limit)
[exact_match] + search_results
end
def exact_match
return unless offset.zero? && username_complete?
return @exact_match if defined?(@exact_match)
@exact_match = begin
if options[:resolve]
ResolveAccountService.new.call(query)
elsif domain_is_local?
Account.find_local(query_username)
else
Account.find_remote(query_username, query_domain)
end
end
end
def resolving_non_matching_remote_account?
options[:resolve] && !exact_match && !domain_is_local?
def search_results
return [] if limit_for_non_exact_results.zero?
@search_results ||= begin
results = from_elasticsearch if Chewy.enabled?
results ||= from_database
results
end
end
def search_results_and_exact_match
exact = [exact_match]
return exact if !exact[0].nil? && limit == 1
exact + search_results.to_a
def from_database
if account
advanced_search_results
else
simple_search_results
end
end
def query_blank_or_hashtag?
query.blank? || query.start_with?('#')
def advanced_search_results
Account.advanced_search_for(terms_for_query, account, limit_for_non_exact_results, options[:following], offset)
end
def simple_search_results
Account.search_for(terms_for_query, limit_for_non_exact_results, offset)
end
def from_elasticsearch
must_clauses = [{ multi_match: { query: terms_for_query, fields: likely_acct? ? %w(acct.edge_ngram acct) : %w(acct.edge_ngram acct display_name.edge_ngram display_name), type: 'most_fields', operator: 'and' } }]
should_clauses = []
if account
return [] if options[:following] && following_ids.empty?
if options[:following]
must_clauses << { terms: { id: following_ids } }
elsif following_ids.any?
should_clauses << { terms: { id: following_ids, boost: 100 } }
end
end
query = { bool: { must: must_clauses, should: should_clauses } }
functions = [reputation_score_function, followers_score_function, time_distance_function]
records = AccountsIndex.query(function_score: { query: query, functions: functions, boost_mode: 'multiply', score_mode: 'avg' })
.limit(limit_for_non_exact_results)
.offset(offset)
.objects
.compact
ActiveRecord::Associations::Preloader.new.preload(records, :account_stat)
records
rescue Faraday::ConnectionFailed, Parslet::ParseFailed
nil
end
def reputation_score_function
{
script_score: {
script: {
source: "(doc['followers_count'].value + 0.0) / (doc['followers_count'].value + doc['following_count'].value + 1)",
},
},
}
end
def followers_score_function
{
field_value_factor: {
field: 'followers_count',
modifier: 'log2p',
missing: 0,
},
}
end
def time_distance_function
{
gauss: {
last_status_at: {
scale: '30d',
offset: '30d',
decay: 0.3,
},
},
}
end
def following_ids
@following_ids ||= account.active_relationships.pluck(:target_account_id)
end
def limit_for_non_exact_results
if exact_match?
limit - 1
else
limit
end
end
def terms_for_query
if domain_is_local?
query_username
else
query
end
end
def split_query_string
@_split_query_string ||= query.gsub(/\A@/, '').split('@')
@split_query_string ||= query.split('@')
end
def query_username
@_query_username ||= split_query_string.first || ''
@query_username ||= split_query_string.first || ''
end
def query_domain
@_query_domain ||= query_without_split? ? nil : split_query_string.last
@query_domain ||= query_without_split? ? nil : split_query_string.last
end
def query_without_split?
@ -56,46 +163,18 @@ class AccountSearchService < BaseService
end
def domain_is_local?
@_domain_is_local ||= TagManager.instance.local_domain?(query_domain)
@domain_is_local ||= TagManager.instance.local_domain?(query_domain)
end
def search_from
options[:following] && account ? account.following : Account
def exact_match?
exact_match.present?
end
def exact_match
@_exact_match ||= begin
if domain_is_local?
search_from.without_suspended.find_local(query_username)
else
search_from.without_suspended.find_remote(query_username, query_domain)
end
end
def username_complete?
query.include?('@') && "@#{query}" =~ Account::MENTION_RE
end
def search_results
@_search_results ||= begin
if account
advanced_search_results
else
simple_search_results
end
end
end
def advanced_search_results
Account.advanced_search_for(terms_for_query, account, limit, options[:following], offset)
end
def simple_search_results
Account.search_for(terms_for_query, limit, offset)
end
def terms_for_query
if domain_is_local?
query_username
else
"#{query_username} #{query_domain}"
end
def likely_acct?
@acct_hint || username_complete?
end
end

3
app/services/activitypub/fetch_featured_collection_service.rb
View File

@ -4,13 +4,12 @@ class ActivityPub::FetchFeaturedCollectionService < BaseService
include JsonLdHelper
def call(account)
return if account.featured_collection_url.blank?
return if account.featured_collection_url.blank? || account.suspended? || account.local?
@account = account
@json = fetch_resource(@account.featured_collection_url, true)
return unless supported_context?
return if @account.suspended? || @account.local?
case @json['type']
when 'Collection', 'CollectionPage'

14
app/services/activitypub/fetch_remote_account_service.rb
View File

@ -2,18 +2,22 @@
class ActivityPub::FetchRemoteAccountService < BaseService
include JsonLdHelper
include DomainControlHelper
SUPPORTED_TYPES = %w(Application Group Organization Person Service).freeze
# Does a WebFinger roundtrip on each call, unless `only_key` is true
def call(uri, id: true, prefetched_body: nil, break_on_redirect: false, only_key: false)
return if domain_not_allowed?(uri)
return ActivityPub::TagManager.instance.uri_to_resource(uri, Account) if ActivityPub::TagManager.instance.local_uri?(uri)
@json = if prefetched_body.nil?
fetch_resource(uri, id)
else
body_to_json(prefetched_body, compare_id: id ? uri : nil)
end
@json = begin
if prefetched_body.nil?
fetch_resource(uri, id)
else
body_to_json(prefetched_body, compare_id: id ? uri : nil)
end
end
return if !supported_context? || !expected_type? || (break_on_redirect && @json['movedTo'].present?)

2
app/services/activitypub/fetch_remote_poll_service.rb
View File

@ -5,7 +5,9 @@ class ActivityPub::FetchRemotePollService < BaseService
def call(poll, on_behalf_of = nil)
json = fetch_resource(poll.status.uri, true, on_behalf_of)
return unless supported_context?(json)
ActivityPub::ProcessPollService.new.call(poll, json)
end
end

20
app/services/activitypub/fetch_remote_status_service.rb
View File

@ -5,18 +5,18 @@ class ActivityPub::FetchRemoteStatusService < BaseService
# Should be called when uri has already been checked for locality
def call(uri, id: true, prefetched_body: nil, on_behalf_of: nil)
@json = if prefetched_body.nil?
fetch_resource(uri, id, on_behalf_of)
else
body_to_json(prefetched_body, compare_id: id ? uri : nil)
end
@json = begin
if prefetched_body.nil?
fetch_resource(uri, id, on_behalf_of)
else
body_to_json(prefetched_body, compare_id: id ? uri : nil)
end
end
return unless supported_context? && expected_type?
return if actor_id.nil? || !trustworthy_attribution?(@json['id'], actor_id)
return if !(supported_context? && expected_type?) || actor_id.nil? || !trustworthy_attribution?(@json['id'], actor_id)
actor = ActivityPub::TagManager.instance.uri_to_resource(actor_id, Account)
actor = ActivityPub::FetchRemoteAccountService.new.call(actor_id, id: true) if actor.nil? || needs_update(actor)
actor = ActivityPub::FetchRemoteAccountService.new.call(actor_id, id: true) if actor.nil? || needs_update?(actor)
return if actor.nil? || actor.suspended?
@ -46,7 +46,7 @@ class ActivityPub::FetchRemoteStatusService < BaseService
equals_or_includes_any?(@json['type'], ActivityPub::Activity::Create::SUPPORTED_TYPES + ActivityPub::Activity::Create::CONVERTED_TYPES)
end
def needs_update(actor)
def needs_update?(actor)
actor.possibly_stale?
end
end

4
app/services/activitypub/process_account_service.rb
View File

@ -2,11 +2,12 @@
class ActivityPub::ProcessAccountService < BaseService
include JsonLdHelper
include DomainControlHelper
# Should be called with confirmed valid JSON
# and WebFinger-resolved username and domain
def call(username, domain, json, options = {})
return if json['inbox'].blank? || unsupported_uri_scheme?(json['id'])
return if json['inbox'].blank? || unsupported_uri_scheme?(json['id']) || domain_not_allowed?(domain)
@options = options
@json = json
@ -84,6 +85,7 @@ class ActivityPub::ProcessAccountService < BaseService
@account.fields = property_values || {}
@account.also_known_as = as_array(@json['alsoKnownAs'] || []).map { |item| value_or_id(item) }
@account.actor_type = actor_type
@account.discoverable = @json['discoverable'] || false
end
def set_fetchable_attributes!

4
app/services/activitypub/process_collection_service.rb
View File

@ -8,9 +8,7 @@ class ActivityPub::ProcessCollectionService < BaseService
@json = Oj.load(body, mode: :strict)
@options = options
return unless supported_context?
return if different_actor? && verify_account!.nil?
return if @account.suspended? || @account.local?
return if !supported_context? || (different_actor? && verify_account!.nil?) || @account.suspended? || @account.local?
case @json['type']
when 'Collection', 'CollectionPage'

6
app/services/activitypub/process_poll_service.rb
View File

@ -5,6 +5,7 @@ class ActivityPub::ProcessPollService < BaseService
def call(poll, json)
@json = json
return unless expected_type?
previous_expires_at = poll.expires_at
@ -27,6 +28,8 @@ class ActivityPub::ProcessPollService < BaseService
end
end
voters_count = @json['votersCount']
latest_options = items.map { |item| item['name'].presence || item['content'] }
# If for some reasons the options were changed, it invalidates all previous
@ -38,7 +41,8 @@ class ActivityPub::ProcessPollService < BaseService
last_fetched_at: Time.now.utc,
expires_at: expires_at,
options: latest_options,
cached_tallies: items.map { |item| item.dig('replies', 'totalItems') || 0 }
cached_tallies: items.map { |item| item.dig('replies', 'totalItems') || 0 },
voters_count: voters_count
)
rescue ActiveRecord::StaleObjectError
poll.reload

7
app/services/app_sign_up_service.rb
View File

@ -4,9 +4,10 @@ class AppSignUpService < BaseService
def call(app, params)
return unless allowed_registrations?
user_params = params.slice(:email, :password, :agreement, :locale)
account_params = params.slice(:username)
user = User.create!(user_params.merge(created_by_application: app, password_confirmation: user_params[:password], account_attributes: account_params))
user_params = params.slice(:email, :password, :agreement, :locale)
account_params = params.slice(:username)
invite_request_params = { text: params[:reason] }
user = User.create!(user_params.merge(created_by_application: app, password_confirmation: user_params[:password], account_attributes: account_params, invite_request_attributes: invite_request_params))
Doorkeeper::AccessToken.create!(application: app,
resource_owner_id: user.id,

12
app/services/authorize_follow_service.rb
View File

@ -11,25 +11,17 @@ class AuthorizeFollowService < BaseService
follow_request.authorize!
end
create_notification(follow_request) unless source_account.local?
create_notification(follow_request) if !source_account.local? && source_account.activitypub?
follow_request
end
private
def create_notification(follow_request)
if follow_request.account.ostatus?
NotificationWorker.perform_async(build_xml(follow_request), follow_request.target_account_id, follow_request.account_id)
elsif follow_request.account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), follow_request.target_account_id, follow_request.account.inbox_url)
end
ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), follow_request.target_account_id, follow_request.account.inbox_url)
end
def build_json(follow_request)
Oj.dump(serialize_payload(follow_request, ActivityPub::AcceptFollowSerializer))
end
def build_xml(follow_request)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.authorize_follow_request_salmon(follow_request))
end
end

42
app/services/batched_remove_status_service.rb
View File

@ -1,7 +1,6 @@
# frozen_string_literal: true
class BatchedRemoveStatusService < BaseService
include StreamEntryRenderer
include Redisable
# Delete given statuses and reblogs of them
@ -9,19 +8,16 @@ class BatchedRemoveStatusService < BaseService
# Dispatch Salmon deletes, unique per domain, of the deleted statuses, but only local ones
# Remove statuses from home feeds
# Push delete events to streaming API for home feeds and public feeds
# @param [Status] statuses A preferably batched array of statuses
# @param [Enumerable<Status>] statuses A preferably batched array of statuses
# @param [Hash] options
# @option [Boolean] :skip_side_effects
def call(statuses, **options)
statuses = Status.where(id: statuses.map(&:id)).includes(:account, :stream_entry).flat_map { |status| [status] + status.reblogs.includes(:account, :stream_entry).to_a }
statuses = Status.where(id: statuses.map(&:id)).includes(:account).flat_map { |status| [status] + status.reblogs.includes(:account).to_a }
@mentions = statuses.each_with_object({}) { |s, h| h[s.id] = s.active_mentions.includes(:account).to_a }
@tags = statuses.each_with_object({}) { |s, h| h[s.id] = s.tags.pluck(:name) }
@stream_entry_batches = []
@salmon_batches = []
@json_payloads = statuses.each_with_object({}) { |s, h| h[s.id] = Oj.dump(event: :delete, payload: s.id.to_s) }
@activity_xml = {}
@json_payloads = statuses.each_with_object({}) { |s, h| h[s.id] = Oj.dump(event: :delete, payload: s.id.to_s) }
# Ensure that rendered XML reflects destroyed state
statuses.each do |status|
@ -39,28 +35,16 @@ class BatchedRemoveStatusService < BaseService
unpush_from_home_timelines(account, account_statuses)
unpush_from_list_timelines(account, account_statuses)
batch_stream_entries(account, account_statuses) if account.local?
end
# Cannot be batched
statuses.each do |status|
unpush_from_public_timelines(status)
batch_salmon_slaps(status) if status.local?
end
Pubsubhubbub::RawDistributionWorker.push_bulk(@stream_entry_batches) { |batch| batch }
NotificationWorker.push_bulk(@salmon_batches) { |batch| batch }
end
private
def batch_stream_entries(account, statuses)
statuses.each do |status|
@stream_entry_batches << [build_xml(status.stream_entry), account.id]
end
end
def unpush_from_home_timelines(account, statuses)
recipients = account.followers_for_local_distribution.to_a
@ -96,25 +80,9 @@ class BatchedRemoveStatusService < BaseService
end
@tags[status.id].each do |hashtag|
redis.publish("timeline:hashtag:#{hashtag}", payload)
redis.publish("timeline:hashtag:#{hashtag}:local", payload) if status.local?
redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", payload)
redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", payload) if status.local?
end
end
end
def batch_salmon_slaps(status)
return if @mentions[status.id].empty?
recipients = @mentions[status.id].map(&:account).reject(&:local?).select(&:ostatus?).uniq(&:domain).map(&:id)
recipients.each do |recipient_id|
@salmon_batches << [build_xml(status.stream_entry), status.account_id, recipient_id]
end
end
def build_xml(stream_entry)
return @activity_xml[stream_entry.id] if @activity_xml.key?(stream_entry.id)
@activity_xml[stream_entry.id] = stream_entry_to_xml(stream_entry)
end
end

14
app/services/block_domain_service.rb
View File

@ -3,13 +3,22 @@
class BlockDomainService < BaseService
attr_reader :domain_block
def call(domain_block)
def call(domain_block, update = false)
@domain_block = domain_block
process_domain_block!
process_retroactive_updates! if update
end
private
def process_retroactive_updates!
# If the domain block severity has been changed, undo the appropriate limitations
scope = Account.by_domain_and_subdomains(domain_block.domain)
scope.where(silenced_at: domain_block.created_at).in_batches.update_all(silenced_at: nil) unless domain_block.silence?
scope.where(suspended_at: domain_block.created_at).in_batches.update_all(suspended_at: nil) unless domain_block.suspend?
end
def process_domain_block!
clear_media! if domain_block.reject_media?
@ -44,8 +53,7 @@ class BlockDomainService < BaseService
def suspend_accounts!
blocked_domain_accounts.without_suspended.reorder(nil).find_each do |account|
UnsubscribeService.new.call(account) if account.subscribed?
SuspendAccountService.new.call(account, suspended_at: @domain_block.created_at)
SuspendAccountService.new.call(account, reserve_username: true, suspended_at: @domain_block.created_at)
end
end

12
app/services/block_service.rb
View File

@ -13,25 +13,17 @@ class BlockService < BaseService
block = account.block!(target_account)
BlockWorker.perform_async(account.id, target_account.id)
create_notification(block) unless target_account.local?
create_notification(block) if !target_account.local? && target_account.activitypub?
block
end
private
def create_notification(block)
if block.target_account.ostatus?
NotificationWorker.perform_async(build_xml(block), block.account_id, block.target_account_id)
elsif block.target_account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(block), block.account_id, block.target_account.inbox_url)
end
ActivityPub::DeliveryWorker.perform_async(build_json(block), block.account_id, block.target_account.inbox_url)
end
def build_json(block)
Oj.dump(serialize_payload(block, ActivityPub::BlockSerializer))
end
def build_xml(block)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.block_salmon(block))
end
end

8
app/services/bootstrap_timeline_service.rb
View File

@ -17,7 +17,11 @@ class BootstrapTimelineService < BaseService
def autofollow_bootstrap_timeline_accounts!
bootstrap_timeline_accounts.each do |target_account|
FollowService.new.call(@source_account, target_account)
begin
FollowService.new.call(@source_account, target_account)
rescue ActiveRecord::RecordNotFound, Mastodon::NotPermittedError
nil
end
end
end
@ -40,7 +44,9 @@ class BootstrapTimelineService < BaseService
def local_unlocked_accounts(usernames)
Account.local
.without_suspended
.where(username: usernames)
.where(locked: false)
.where(moved_to_account_id: nil)
end
end

23
app/services/concerns/author_extractor.rb
View File

@ -1,23 +0,0 @@
# frozen_string_literal: true
module AuthorExtractor
def author_from_xml(xml, update_profile = true)
return nil if xml.nil?
# Try <email> for acct
acct = xml.at_xpath('./xmlns:author/xmlns:email', xmlns: OStatus::TagManager::XMLNS)&.content
# Try <name> + <uri>
if acct.blank?
username = xml.at_xpath('./xmlns:author/xmlns:name', xmlns: OStatus::TagManager::XMLNS)&.content
uri = xml.at_xpath('./xmlns:author/xmlns:uri', xmlns: OStatus::TagManager::XMLNS)&.content
return nil if username.blank? || uri.blank?
domain = Addressable::URI.parse(uri).normalized_host
acct = "#{username}@#{domain}"
end
ResolveAccountService.new.call(acct, update_profile: update_profile)
end
end

2
app/services/concerns/payloadable.rb
View File

@ -14,6 +14,6 @@ module Payloadable
end
def signing_enabled?
true
ENV['AUTHORIZED_FETCH'] != 'true' && !Rails.configuration.x.whitelist_mode
end
end

7
app/services/concerns/stream_entry_renderer.rb
View File

@ -1,7 +0,0 @@
# frozen_string_literal: true
module StreamEntryRenderer
def stream_entry_to_xml(stream_entry)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.entry(stream_entry, true))
end
end

4
app/services/fan_out_on_write_service.rb
View File

@ -72,8 +72,8 @@ class FanOutOnWriteService < BaseService
Rails.logger.debug "Delivering status #{status.id} to hashtags"
status.tags.pluck(:name).each do |hashtag|
Redis.current.publish("timeline:hashtag:#{hashtag}", @payload)
Redis.current.publish("timeline:hashtag:#{hashtag}:local", @payload) if status.local?
Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", @payload)
Redis.current.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", @payload) if status.local?
end
end

6
app/services/favourite_service.rb
View File

@ -30,8 +30,6 @@ class FavouriteService < BaseService
if status.account.local?
NotifyService.new.call(status.account, favourite)
elsif status.account.ostatus?
NotificationWorker.perform_async(build_xml(favourite), favourite.account_id, status.account_id)
elsif status.account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(favourite), favourite.account_id, status.account.inbox_url)
end
@ -46,8 +44,4 @@ class FavouriteService < BaseService
def build_json(favourite)
Oj.dump(serialize_payload(favourite, ActivityPub::LikeSerializer))
end
def build_xml(favourite)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.favourite_salmon(favourite))
end
end

93
app/services/fetch_atom_service.rb
View File

@ -1,93 +0,0 @@
# frozen_string_literal: true
class FetchAtomService < BaseService
include JsonLdHelper
def call(url)
return if url.blank?
result = process(url)
# retry without ActivityPub
result ||= process(url) if @unsupported_activity
result
rescue OpenSSL::SSL::SSLError => e
Rails.logger.debug "SSL error: #{e}"
nil
rescue HTTP::ConnectionError => e
Rails.logger.debug "HTTP ConnectionError: #{e}"
nil
end
private
def process(url, terminal = false)
@url = url
perform_request { |response| process_response(response, terminal) }
end
def perform_request(&block)
accept = 'text/html'
accept = 'application/activity+json, application/ld+json; profile="https://www.w3.org/ns/activitystreams", application/atom+xml, ' + accept unless @unsupported_activity
Request.new(:get, @url).add_headers('Accept' => accept).perform(&block)
end
def process_response(response, terminal = false)
return nil if response.code != 200
if response.mime_type == 'application/atom+xml'
[@url, { prefetched_body: response.body_with_limit }, :ostatus]
elsif ['application/activity+json', 'application/ld+json'].include?(response.mime_type)
body = response.body_with_limit
json = body_to_json(body)
if supported_context?(json) && equals_or_includes_any?(json['type'], ActivityPub::FetchRemoteAccountService::SUPPORTED_TYPES) && json['inbox'].present?
[json['id'], { prefetched_body: body, id: true }, :activitypub]
elsif supported_context?(json) && expected_type?(json)
[json['id'], { prefetched_body: body, id: true }, :activitypub]
else
@unsupported_activity = true
nil
end
elsif !terminal
link_header = response['Link'] && parse_link_header(response)
if link_header&.find_link(%w(rel alternate))
process_link_headers(link_header)
elsif response.mime_type == 'text/html'
process_html(response)
end
end
end
def expected_type?(json)
equals_or_includes_any?(json['type'], ActivityPub::Activity::Create::SUPPORTED_TYPES + ActivityPub::Activity::Create::CONVERTED_TYPES)
end
def process_html(response)
page = Nokogiri::HTML(response.body_with_limit)
json_link = page.xpath('//link[@rel="alternate"]').find { |link| ['application/activity+json', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'].include?(link['type']) }
atom_link = page.xpath('//link[@rel="alternate"]').find { |link| link['type'] == 'application/atom+xml' }
result ||= process(json_link['href'], terminal: true) unless json_link.nil? || @unsupported_activity
result ||= process(atom_link['href'], terminal: true) unless atom_link.nil?
result
end
def process_link_headers(link_header)
json_link = link_header.find_link(%w(rel alternate), %w(type application/activity+json)) || link_header.find_link(%w(rel alternate), ['type', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'])
atom_link = link_header.find_link(%w(rel alternate), %w(type application/atom+xml))
result ||= process(json_link.href, terminal: true) unless json_link.nil? || @unsupported_activity
result ||= process(atom_link.href, terminal: true) unless atom_link.nil?
result
end
def parse_link_header(response)
LinkHeader.parse(response['Link'].is_a?(Array) ? response['Link'].first : response['Link'])
end
end

12
app/services/fetch_link_card_service.rb
View File

@ -22,14 +22,14 @@ class FetchLinkCardService < BaseService
RedisLock.acquire(lock_options) do |lock|
if lock.acquired?
@card = PreviewCard.find_by(url: @url)
process_url if @card.nil? || @card.updated_at <= 2.weeks.ago
process_url if @card.nil? || @card.updated_at <= 2.weeks.ago || @card.missing_image?
else
raise Mastodon::RaceConditionError
end
end
attach_card if @card&.persisted?
rescue HTTP::Error, Addressable::URI::InvalidURIError, Mastodon::HostValidationError, Mastodon::LengthValidationError => e
rescue HTTP::Error, OpenSSL::SSL::SSLError, Addressable::URI::InvalidURIError, Mastodon::HostValidationError, Mastodon::LengthValidationError => e
Rails.logger.debug "Error fetching link #{@url}: #{e}"
nil
end
@ -39,12 +39,6 @@ class FetchLinkCardService < BaseService
def process_url
@card ||= PreviewCard.new(url: @url)
failed = Request.new(:head, @url).perform do |res|
res.code != 405 && res.code != 501 && (res.code != 200 || res.mime_type != 'text/html')
end
return if failed
Request.new(:get, @url).perform do |res|
if res.code == 200 && res.mime_type == 'text/html'
@html = res.body_with_limit
@ -84,7 +78,7 @@ class FetchLinkCardService < BaseService
def mention_link?(a)
@status.mentions.any? do |mention|
a['href'] == TagManager.instance.url_for(mention.account)
a['href'] == ActivityPub::TagManager.instance.url_for(mention.account)
end
end

30
app/services/fetch_remote_account_service.rb
View File

@ -1,45 +1,17 @@
# frozen_string_literal: true
class FetchRemoteAccountService < BaseService
include AuthorExtractor
def call(url, prefetched_body = nil, protocol = :ostatus)
if prefetched_body.nil?
resource_url, resource_options, protocol = FetchAtomService.new.call(url)
resource_url, resource_options, protocol = FetchResourceService.new.call(url)
else
resource_url = url
resource_options = { prefetched_body: prefetched_body }
end
case protocol
when :ostatus
process_atom(resource_url, **resource_options)
when :activitypub
ActivityPub::FetchRemoteAccountService.new.call(resource_url, **resource_options)
end
end
private
def process_atom(url, prefetched_body:)
xml = Nokogiri::XML(prefetched_body)
xml.encoding = 'utf-8'
account = author_from_xml(xml.at_xpath('/xmlns:feed', xmlns: OStatus::TagManager::XMLNS), false)
UpdateRemoteProfileService.new.call(xml, account) if account.present? && trusted_domain?(url, account)
account
rescue TypeError
Rails.logger.debug "Unparseable URL given: #{url}"
nil
rescue Nokogiri::XML::XPath::SyntaxError
Rails.logger.debug 'Invalid XML or missing namespace'
nil
end
def trusted_domain?(url, account)
domain = Addressable::URI.parse(url).normalized_host
domain.casecmp(account.domain).zero? || domain.casecmp(Addressable::URI.parse(account.remote_url.presence || account.uri).normalized_host).zero?
end
end

30
app/services/fetch_remote_status_service.rb
View File

@ -1,45 +1,17 @@
# frozen_string_literal: true
class FetchRemoteStatusService < BaseService
include AuthorExtractor
def call(url, prefetched_body = nil, protocol = :ostatus)
if prefetched_body.nil?
resource_url, resource_options, protocol = FetchAtomService.new.call(url)
resource_url, resource_options, protocol = FetchResourceService.new.call(url)
else
resource_url = url
resource_options = { prefetched_body: prefetched_body }
end
case protocol
when :ostatus
process_atom(resource_url, **resource_options)
when :activitypub
ActivityPub::FetchRemoteStatusService.new.call(resource_url, **resource_options)
end
end
private
def process_atom(url, prefetched_body:)
Rails.logger.debug "Processing Atom for remote status at #{url}"
xml = Nokogiri::XML(prefetched_body)
xml.encoding = 'utf-8'
account = author_from_xml(xml.at_xpath('/xmlns:entry', xmlns: OStatus::TagManager::XMLNS))
domain = Addressable::URI.parse(url).normalized_host
return nil unless !account.nil? && confirmed_domain?(domain, account)
statuses = ProcessFeedService.new.call(prefetched_body, account)
statuses.first
rescue Nokogiri::XML::XPath::SyntaxError
Rails.logger.debug 'Invalid XML or missing namespace'
nil
end
def confirmed_domain?(domain, account)
account.domain.nil? || domain.casecmp(account.domain).zero? || domain.casecmp(Addressable::URI.parse(account.remote_url.presence || account.uri).normalized_host).zero?
end
end

68
app/services/fetch_resource_service.rb Normal file
View File

@ -0,0 +1,68 @@
# frozen_string_literal: true
class FetchResourceService < BaseService
include JsonLdHelper
ACCEPT_HEADER = 'application/activity+json, application/ld+json; profile="https://www.w3.org/ns/activitystreams", text/html'
def call(url)
return if url.blank?
process(url)
rescue HTTP::Error, OpenSSL::SSL::SSLError, Addressable::URI::InvalidURIError, Mastodon::HostValidationError, Mastodon::LengthValidationError => e
Rails.logger.debug "Error fetching resource #{@url}: #{e}"
nil
end
private
def process(url, terminal = false)
@url = url
perform_request { |response| process_response(response, terminal) }
end
def perform_request(&block)
Request.new(:get, @url).add_headers('Accept' => ACCEPT_HEADER).on_behalf_of(Account.representative).perform(&block)
end
def process_response(response, terminal = false)
return nil if response.code != 200
if ['application/activity+json', 'application/ld+json'].include?(response.mime_type)
body = response.body_with_limit
json = body_to_json(body)
[json['id'], { prefetched_body: body, id: true }, :activitypub] if supported_context?(json) && (equals_or_includes_any?(json['type'], ActivityPub::FetchRemoteAccountService::SUPPORTED_TYPES) || expected_type?(json))
elsif !terminal
link_header = response['Link'] && parse_link_header(response)
if link_header&.find_link(%w(rel alternate))
process_link_headers(link_header)
elsif response.mime_type == 'text/html'
process_html(response)
end
end
end
def expected_type?(json)
equals_or_includes_any?(json['type'], ActivityPub::Activity::Create::SUPPORTED_TYPES + ActivityPub::Activity::Create::CONVERTED_TYPES)
end
def process_html(response)
page = Nokogiri::HTML(response.body_with_limit)
json_link = page.xpath('//link[@rel="alternate"]').find { |link| ['application/activity+json', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'].include?(link['type']) }
process(json_link['href'], terminal: true) unless json_link.nil?
end
def process_link_headers(link_header)
json_link = link_header.find_link(%w(rel alternate), %w(type application/activity+json)) || link_header.find_link(%w(rel alternate), ['type', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'])
process(json_link.href, terminal: true) unless json_link.nil?
end
def parse_link_header(response)
LinkHeader.parse(response['Link'].is_a?(Array) ? response['Link'].first : response['Link'])
end
end

26
app/services/follow_service.rb
View File

@ -13,7 +13,7 @@ class FollowService < BaseService
target_account = ResolveAccountService.new.call(target_account, skip_webfinger: true)
raise ActiveRecord::RecordNotFound if target_account.nil? || target_account.id == source_account.id || target_account.suspended?
raise Mastodon::NotPermittedError if target_account.blocking?(source_account) || source_account.blocking?(target_account) || target_account.moved? || source_account.domain_blocking?(target_account.domain)
raise Mastodon::NotPermittedError if target_account.blocking?(source_account) || source_account.blocking?(target_account) || target_account.moved? || (!target_account.local? && target_account.ostatus?) || source_account.domain_blocking?(target_account.domain)
if source_account.following?(target_account)
# We're already following this account, but we'll call follow! again to
@ -30,9 +30,9 @@ class FollowService < BaseService
ActivityTracker.increment('activity:interactions')
if target_account.locked? || target_account.activitypub?
if target_account.locked? || source_account.silenced? || target_account.activitypub?
request_follow(source_account, target_account, reblogs: reblogs)
else
elsif target_account.local?
direct_follow(source_account, target_account, reblogs: reblogs)
end
end
@ -44,9 +44,6 @@ class FollowService < BaseService
if target_account.local?
LocalNotificationWorker.perform_async(target_account.id, follow_request.id, follow_request.class.name)
elsif target_account.ostatus?
NotificationWorker.perform_async(build_follow_request_xml(follow_request), source_account.id, target_account.id)
AfterRemoteFollowRequestWorker.perform_async(follow_request.id)
elsif target_account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), source_account.id, target_account.inbox_url)
end
@ -57,27 +54,12 @@ class FollowService < BaseService
def direct_follow(source_account, target_account, reblogs: true)
follow = source_account.follow!(target_account, reblogs: reblogs)
if target_account.local?
LocalNotificationWorker.perform_async(target_account.id, follow.id, follow.class.name)
else
Pubsubhubbub::SubscribeWorker.perform_async(target_account.id) unless target_account.subscribed?
NotificationWorker.perform_async(build_follow_xml(follow), source_account.id, target_account.id)
AfterRemoteFollowWorker.perform_async(follow.id)
end
LocalNotificationWorker.perform_async(target_account.id, follow.id, follow.class.name)
MergeWorker.perform_async(target_account.id, source_account.id)
follow
end
def build_follow_request_xml(follow_request)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.follow_request_salmon(follow_request))
end
def build_follow_xml(follow)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.follow_salmon(follow))
end
def build_json(follow_request)
Oj.dump(serialize_payload(follow_request, ActivityPub::FollowSerializer))
end

4
app/services/hashtag_query_service.rb
View File

@ -14,7 +14,7 @@ class HashtagQueryService < BaseService
private
def tags_for(tags)
Tag.where(name: tags.map(&:downcase)) if tags.presence
def tags_for(names)
Tag.matching_name(names) if names.presence
end
end

32
app/services/move_service.rb Normal file
View File

@ -0,0 +1,32 @@
# frozen_string_literal: true
class MoveService < BaseService
def call(migration)
@migration = migration
@source_account = migration.account
@target_account = migration.target_account
update_redirect!
process_local_relationships!
distribute_update!
distribute_move!
end
private
def update_redirect!
@source_account.update!(moved_to_account: @target_account)
end
def process_local_relationships!
MoveWorker.perform_async(@source_account.id, @target_account.id)
end
def distribute_update!
ActivityPub::UpdateDistributionWorker.perform_async(@source_account.id)
end
def distribute_move!
ActivityPub::MoveDistributionWorker.perform_async(@migration.id)
end
end

3
app/services/post_status_service.rb
View File

@ -95,7 +95,6 @@ class PostStatusService < BaseService
LinkCrawlWorker.perform_async(@status.id) unless @status.spoiler_text?
DistributionWorker.perform_async(@status.id)
unless @status.local_only?
Pubsubhubbub::DistributionWorker.perform_async(@status.stream_entry.id)
ActivityPub::DistributionWorker.perform_async(@status.id)
PollExpirationNotifyWorker.perform_at(@status.poll.expires_at, @status.poll.id) if @status.poll
end
@ -184,7 +183,7 @@ class PostStatusService < BaseService
def poll_attributes
return if @options[:poll].blank?
@options[:poll].merge(account: @account)
@options[:poll].merge(account: @account, voters_count: 0)
end
def scheduled_options

31
app/services/process_feed_service.rb
View File

@ -1,31 +0,0 @@
# frozen_string_literal: true
class ProcessFeedService < BaseService
def call(body, account, **options)
@options = options
xml = Nokogiri::XML(body)
xml.encoding = 'utf-8'
update_author(body, account)
process_entries(xml, account)
end
private
def update_author(body, account)
RemoteProfileUpdateWorker.perform_async(account.id, body.force_encoding('UTF-8'), true)
end
def process_entries(xml, account)
xml.xpath('//xmlns:entry', xmlns: OStatus::TagManager::XMLNS).reverse_each.map { |entry| process_entry(entry, account) }.compact
end
def process_entry(xml, account)
activity = OStatus::Activity::General.new(xml, account, @options)
activity.specialize&.perform if activity.status?
rescue ActiveRecord::RecordInvalid => e
Rails.logger.debug "Nothing was saved for #{activity.id} because: #{e}"
nil
end
end

6
app/services/process_hashtags_service.rb
View File

@ -5,16 +5,14 @@ class ProcessHashtagsService < BaseService
tags = Extractor.extract_hashtags(status.text) if status.local?
records = []
tags.map { |str| str.mb_chars.downcase }.uniq(&:to_s).each do |name|
tag = Tag.where(name: name).first_or_create(name: name)
Tag.find_or_create_by_names(tags) do |tag|
status.tags << tag
records << tag
TrendingTags.record_use!(tag, status.account, status.created_at) if status.public_visibility?
end
return unless status.public_visibility? || status.unlisted_visibility?
return unless status.distributable?
status.account.featured_tags.where(tag_id: records.map(&:id)).each do |featured_tag|
featured_tag.increment(status.created_at)

151
app/services/process_interaction_service.rb
View File

@ -1,151 +0,0 @@
# frozen_string_literal: true
class ProcessInteractionService < BaseService
include AuthorExtractor
include Authorization
# Record locally the remote interaction with our user
# @param [String] envelope Salmon envelope
# @param [Account] target_account Account the Salmon was addressed to
def call(envelope, target_account)
body = salmon.unpack(envelope)
xml = Nokogiri::XML(body)
xml.encoding = 'utf-8'
account = author_from_xml(xml.at_xpath('/xmlns:entry', xmlns: OStatus::TagManager::XMLNS))
return if account.nil? || account.suspended?
if salmon.verify(envelope, account.keypair)
RemoteProfileUpdateWorker.perform_async(account.id, body.force_encoding('UTF-8'), true)
case verb(xml)
when :follow
follow!(account, target_account) unless target_account.locked? || target_account.blocking?(account) || target_account.domain_blocking?(account.domain)
when :request_friend
follow_request!(account, target_account) unless !target_account.locked? || target_account.blocking?(account) || target_account.domain_blocking?(account.domain)
when :authorize
authorize_follow_request!(account, target_account)
when :reject
reject_follow_request!(account, target_account)
when :unfollow
unfollow!(account, target_account)
when :favorite
favourite!(xml, account)
when :unfavorite
unfavourite!(xml, account)
when :post
add_post!(body, account) if mentions_account?(xml, target_account)
when :share
add_post!(body, account) unless status(xml).nil?
when :delete
delete_post!(xml, account)
when :block
reflect_block!(account, target_account)
when :unblock
reflect_unblock!(account, target_account)
end
end
rescue HTTP::Error, OStatus2::BadSalmonError, Mastodon::NotPermittedError
nil
end
private
def mentions_account?(xml, account)
xml.xpath('/xmlns:entry/xmlns:link[@rel="mentioned"]', xmlns: OStatus::TagManager::XMLNS).each { |mention_link| return true if [OStatus::TagManager.instance.uri_for(account), OStatus::TagManager.instance.url_for(account)].include?(mention_link.attribute('href').value) }
false
end
def verb(xml)
raw = xml.at_xpath('//activity:verb', activity: OStatus::TagManager::AS_XMLNS).content
OStatus::TagManager::VERBS.key(raw)
rescue
:post
end
def follow!(account, target_account)
follow = account.follow!(target_account)
FollowRequest.find_by(account: account, target_account: target_account)&.destroy
NotifyService.new.call(target_account, follow)
end
def follow_request!(account, target_account)
return if account.requested?(target_account)
follow_request = FollowRequest.create!(account: account, target_account: target_account)
NotifyService.new.call(target_account, follow_request)
end
def authorize_follow_request!(account, target_account)
follow_request = FollowRequest.find_by(account: target_account, target_account: account)
follow_request&.authorize!
Pubsubhubbub::SubscribeWorker.perform_async(account.id) unless account.subscribed?
end
def reject_follow_request!(account, target_account)
follow_request = FollowRequest.find_by(account: target_account, target_account: account)
follow_request&.reject!
end
def unfollow!(account, target_account)
account.unfollow!(target_account)
FollowRequest.find_by(account: account, target_account: target_account)&.destroy
end
def reflect_block!(account, target_account)
UnfollowService.new.call(target_account, account) if target_account.following?(account)
account.block!(target_account)
end
def reflect_unblock!(account, target_account)
UnblockService.new.call(account, target_account)
end
def delete_post!(xml, account)
status = Status.find(xml.at_xpath('//xmlns:id', xmlns: OStatus::TagManager::XMLNS).content)
return if status.nil?
authorize_with account, status, :destroy?
RemovalWorker.perform_async(status.id)
end
def favourite!(xml, from_account)
current_status = status(xml)
return if current_status.nil?
favourite = current_status.favourites.where(account: from_account).first_or_create!(account: from_account)
NotifyService.new.call(current_status.account, favourite)
end
def unfavourite!(xml, from_account)
current_status = status(xml)
return if current_status.nil?
favourite = current_status.favourites.where(account: from_account).first
favourite&.destroy
end
def add_post!(body, account)
ProcessingWorker.perform_async(account.id, body.force_encoding('UTF-8'))
end
def status(xml)
uri = activity_id(xml)
return nil unless OStatus::TagManager.instance.local_id?(uri)
Status.find(OStatus::TagManager.instance.unique_tag_to_local_id(uri, 'Status'))
end
def activity_id(xml)
xml.at_xpath('//activity:object', activity: OStatus::TagManager::AS_XMLNS).at_xpath('./xmlns:id', xmlns: OStatus::TagManager::XMLNS).content
end
def salmon
@salmon ||= OStatus2::Salmon.new
end
end

14
app/services/process_mentions_service.rb
View File

@ -1,7 +1,6 @@
# frozen_string_literal: true
class ProcessMentionsService < BaseService
include StreamEntryRenderer
include Payloadable
# Scan status for mentions and fetch remote mentioned users, create
@ -34,6 +33,7 @@ class ProcessMentionsService < BaseService
end
status.save!
check_for_spam(status)
mentions.each { |mention| create_notification(mention) }
end
@ -41,7 +41,7 @@ class ProcessMentionsService < BaseService
private
def mention_undeliverable?(mentioned_account)
mentioned_account.nil? || (!mentioned_account.local? && mentioned_account.ostatus? && @status.stream_entry.hidden?)
mentioned_account.nil? || (!mentioned_account.local? && mentioned_account.ostatus?)
end
def create_notification(mention)
@ -49,17 +49,11 @@ class ProcessMentionsService < BaseService
if mentioned_account.local?
LocalNotificationWorker.perform_async(mentioned_account.id, mention.id, mention.class.name)
elsif mentioned_account.ostatus? && !@status.stream_entry.hidden? && !@status.local_only?
NotificationWorker.perform_async(ostatus_xml, @status.account_id, mentioned_account.id)
elsif mentioned_account.activitypub? && !@status.local_only?
ActivityPub::DeliveryWorker.perform_async(activitypub_json, mention.status.account_id, mentioned_account.inbox_url)
end
end
def ostatus_xml
@ostatus_xml ||= stream_entry_to_xml(@status.stream_entry)
end
def activitypub_json
return @activitypub_json if defined?(@activitypub_json)
@activitypub_json = Oj.dump(serialize_payload(@status, ActivityPub::ActivitySerializer, signer: @status.account))
@ -68,4 +62,8 @@ class ProcessMentionsService < BaseService
def resolve_account_service
ResolveAccountService.new
end
def check_for_spam(status)
SpamCheck.perform(status)
end
end

53
app/services/pubsubhubbub/subscribe_service.rb
View File

@ -1,53 +0,0 @@
# frozen_string_literal: true
class Pubsubhubbub::SubscribeService < BaseService
URL_PATTERN = /\A#{URI.regexp(%w(http https))}\z/
attr_reader :account, :callback, :secret,
:lease_seconds, :domain
def call(account, callback, secret, lease_seconds, verified_domain = nil)
@account = account
@callback = Addressable::URI.parse(callback).normalize.to_s
@secret = secret
@lease_seconds = lease_seconds
@domain = verified_domain
process_subscribe
end
private
def process_subscribe
if account.nil?
['Invalid topic URL', 422]
elsif !valid_callback?
['Invalid callback URL', 422]
elsif blocked_domain?
['Callback URL not allowed', 403]
else
confirm_subscription
['', 202]
end
end
def confirm_subscription
subscription = locate_subscription
Pubsubhubbub::ConfirmationWorker.perform_async(subscription.id, 'subscribe', secret, lease_seconds)
end
def valid_callback?
callback.present? && callback =~ URL_PATTERN
end
def blocked_domain?
DomainBlock.blocked? Addressable::URI.parse(callback).host
end
def locate_subscription
subscription = Subscription.find_or_initialize_by(account: account, callback_url: callback)
subscription.domain = domain
subscription.save!
subscription
end
end

31
app/services/pubsubhubbub/unsubscribe_service.rb
View File

@ -1,31 +0,0 @@
# frozen_string_literal: true
class Pubsubhubbub::UnsubscribeService < BaseService
attr_reader :account, :callback
def call(account, callback)
@account = account
@callback = Addressable::URI.parse(callback).normalize.to_s
process_unsubscribe
end
private
def process_unsubscribe
if account.nil?
['Invalid topic URL', 422]
else
confirm_unsubscribe unless subscription.nil?
['', 202]
end
end
def confirm_unsubscribe
Pubsubhubbub::ConfirmationWorker.perform_async(subscription.id, 'unsubscribe')
end
def subscription
@_subscription ||= Subscription.find_by(account: account, callback_url: callback)
end
end

9
app/services/reblog_service.rb
View File

@ -2,7 +2,6 @@
class ReblogService < BaseService
include Authorization
include StreamEntryRenderer
include Payloadable
# Reblog a status and notify its remote author
@ -24,11 +23,7 @@ class ReblogService < BaseService
reblog = account.statuses.create!(reblog: reblogged_status, text: '', visibility: visibility)
DistributionWorker.perform_async(reblog.id)
unless reblogged_status.local_only?
Pubsubhubbub::DistributionWorker.perform_async(reblog.stream_entry.id)
ActivityPub::DistributionWorker.perform_async(reblog.id)
end
ActivityPub::DistributionWorker.perform_async(reblog.id)
create_notification(reblog)
bump_potential_friendship(account, reblog)
@ -43,8 +38,6 @@ class ReblogService < BaseService
if reblogged_status.account.local?
LocalNotificationWorker.perform_async(reblogged_status.account_id, reblog.id, reblog.class.name)
elsif reblogged_status.account.ostatus?
NotificationWorker.perform_async(stream_entry_to_xml(reblog.stream_entry), reblog.account_id, reblogged_status.account_id)
elsif reblogged_status.account.activitypub? && !reblogged_status.account.following?(reblog.account)
ActivityPub::DeliveryWorker.perform_async(build_json(reblog), reblog.account_id, reblogged_status.account.inbox_url)
end

12
app/services/reject_follow_service.rb
View File

@ -6,25 +6,17 @@ class RejectFollowService < BaseService
def call(source_account, target_account)
follow_request = FollowRequest.find_by!(account: source_account, target_account: target_account)
follow_request.reject!
create_notification(follow_request) unless source_account.local?
create_notification(follow_request) if !source_account.local? && source_account.activitypub?
follow_request
end
private
def create_notification(follow_request)
if follow_request.account.ostatus?
NotificationWorker.perform_async(build_xml(follow_request), follow_request.target_account_id, follow_request.account_id)
elsif follow_request.account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), follow_request.target_account_id, follow_request.account.inbox_url)
end
ActivityPub::DeliveryWorker.perform_async(build_json(follow_request), follow_request.target_account_id, follow_request.account.inbox_url)
end
def build_json(follow_request)
Oj.dump(serialize_payload(follow_request, ActivityPub::RejectFollowSerializer))
end
def build_xml(follow_request)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.reject_follow_request_salmon(follow_request))
end
end

52
app/services/remove_status_service.rb
View File

@ -1,19 +1,23 @@
# frozen_string_literal: true
class RemoveStatusService < BaseService
include StreamEntryRenderer
include Redisable
include Payloadable
# Delete a status
# @param [Status] status
# @param [Hash] options
# @option [Boolean] :redraft
# @option [Boolean] :immediate
# @option [Boolean] :original_removed
def call(status, **options)
@payload = Oj.dump(event: :delete, payload: status.id.to_s)
@status = status
@account = status.account
@tags = status.tags.pluck(:name).to_a
@mentions = status.active_mentions.includes(:account).to_a
@reblogs = status.reblogs.includes(:account).to_a
@stream_entry = status.stream_entry
@options = options
@payload = Oj.dump(event: :delete, payload: status.id.to_s)
@status = status
@account = status.account
@tags = status.tags.pluck(:name).to_a
@mentions = status.active_mentions.includes(:account).to_a
@reblogs = status.reblogs.includes(:account).to_a
@options = options
RedisLock.acquire(lock_options) do |lock|
if lock.acquired?
@ -25,8 +29,10 @@ class RemoveStatusService < BaseService
remove_from_hashtags
remove_from_public
remove_from_media if status.media_attachments.any?
remove_from_spam_check
remove_media
@status.destroy!
@status.destroy! if @options[:immediate] || !@status.reported?
else
raise Mastodon::RaceConditionError
end
@ -78,11 +84,6 @@ class RemoveStatusService < BaseService
target_accounts << @status.reblog.account if @status.reblog? && !@status.reblog.account.local?
target_accounts.uniq!(&:id)
# Ostatus
NotificationWorker.push_bulk(target_accounts.select(&:ostatus?).uniq(&:domain)) do |target_account|
[salmon_xml, @account.id, target_account.id]
end
# ActivityPub
ActivityPub::DeliveryWorker.push_bulk(target_accounts.select(&:activitypub?).uniq(&:preferred_inbox_url)) do |target_account|
[signed_activity_json, @account.id, target_account.preferred_inbox_url]
@ -90,9 +91,6 @@ class RemoveStatusService < BaseService
end
def remove_from_remote_followers
# OStatus
Pubsubhubbub::RawDistributionWorker.perform_async(salmon_xml, @account.id)
# ActivityPub
ActivityPub::DeliveryWorker.push_bulk(@account.followers.inboxes) do |inbox_url|
[signed_activity_json, @account.id, inbox_url]
@ -111,10 +109,6 @@ class RemoveStatusService < BaseService
end
end
def salmon_xml
@salmon_xml ||= stream_entry_to_xml(@stream_entry)
end
def signed_activity_json
@signed_activity_json ||= Oj.dump(serialize_payload(@status, @status.reblog? ? ActivityPub::UndoAnnounceSerializer : ActivityPub::DeleteSerializer, signer: @account))
end
@ -137,8 +131,8 @@ class RemoveStatusService < BaseService
return unless @status.public_visibility?
@tags.each do |hashtag|
redis.publish("timeline:hashtag:#{hashtag}", @payload)
redis.publish("timeline:hashtag:#{hashtag}:local", @payload) if @status.local?
redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", @payload)
redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", @payload) if @status.local?
end
end
@ -156,6 +150,16 @@ class RemoveStatusService < BaseService
redis.publish('timeline:public:local:media', @payload) if @status.local?
end
def remove_media
return if @options[:redraft] || (!@options[:immediate] && @status.reported?)
@status.media_attachments.destroy_all
end
def remove_from_spam_check
redis.zremrangebyscore("spam_check:#{@status.account_id}", @status.id, @status.id)
end
def lock_options
{ redis: Redis.current, key: "distribute:#{@status.id}" }
end

227
app/services/resolve_account_service.rb
View File

@ -1,89 +1,113 @@
# frozen_string_literal: true
class ResolveAccountService < BaseService
include OStatus2::MagicKey
include JsonLdHelper
include DomainControlHelper
DFRN_NS = 'http://purl.org/macgirvin/dfrn/1.0'
class WebfingerRedirectError < StandardError; end
# Find or create a local account for a remote user.
# When creating, look up the user's webfinger and fetch all
# important information from their feed
# @param [String, Account] uri User URI in the form of username@domain
# Find or create an account record for a remote user. When creating,
# look up the user's webfinger and fetch ActivityPub data
# @param [String, Account] uri URI in the username@domain format or account record
# @param [Hash] options
# @option options [Boolean] :redirected Do not follow further Webfinger redirects
# @option options [Boolean] :skip_webfinger Do not attempt to refresh account data
# @return [Account]
def call(uri, options = {})
return if uri.blank?
process_options!(uri, options)
# First of all we want to check if we've got the account
# record with the URI already, and if so, we can exit early
return if domain_not_allowed?(@domain)
@account ||= Account.find_remote(@username, @domain)
return @account if @account&.local? || !webfinger_update_due?
# At this point we are in need of a Webfinger query, which may
# yield us a different username/domain through a redirect
process_webfinger!(@uri)
# Because the username/domain pair may be different than what
# we already checked, we need to check if we've already got
# the record with that URI, again
return if domain_not_allowed?(@domain)
@account ||= Account.find_remote(@username, @domain)
return @account if @account&.local? || !webfinger_update_due?
# Now it is certain, it is definitely a remote account, and it
# either needs to be created, or updated from fresh data
process_account!
rescue Goldfinger::Error, WebfingerRedirectError, Oj::ParseError => e
Rails.logger.debug "Webfinger query for #{@uri} failed: #{e}"
nil
end
private
def process_options!(uri, options)
@options = options
if uri.is_a?(Account)
@account = uri
@username = @account.username
@domain = @account.domain
uri = "#{@username}@#{@domain}"
return @account if @account.local? || !webfinger_update_due?
else
@username, @domain = uri.split('@')
return Account.find_local(@username) if TagManager.instance.local_domain?(@domain)
@account = Account.find_remote(@username, @domain)
return @account unless webfinger_update_due?
end
Rails.logger.debug "Looking up webfinger for #{uri}"
@domain = begin
if TagManager.instance.local_domain?(@domain)
nil
else
TagManager.instance.normalize_domain(@domain)
end
end
@webfinger = Goldfinger.finger("acct:#{uri}")
@uri = [@username, @domain].compact.join('@')
end
def process_webfinger!(uri, redirected = false)
@webfinger = Goldfinger.finger("acct:#{uri}")
confirmed_username, confirmed_domain = @webfinger.subject.gsub(/\Aacct:/, '').split('@')
if confirmed_username.casecmp(@username).zero? && confirmed_domain.casecmp(@domain).zero?
@username = confirmed_username
@domain = confirmed_domain
elsif options[:redirected].nil?
return call("#{confirmed_username}@#{confirmed_domain}", options.merge(redirected: true))
@uri = uri
elsif !redirected
return process_webfinger!("#{confirmed_username}@#{confirmed_domain}", true)
else
Rails.logger.debug 'Requested and returned acct URIs do not match'
return
raise WebfingerRedirectError, "The URI #{uri} tries to hijack #{@username}@#{@domain}"
end
return if links_missing? || auto_suspend?
return Account.find_local(@username) if TagManager.instance.local_domain?(@domain)
@domain = nil if TagManager.instance.local_domain?(@domain)
end
def process_account!
return unless activitypub_ready?
RedisLock.acquire(lock_options) do |lock|
if lock.acquired?
@account = Account.find_remote(@username, @domain)
if activitypub_ready? || @account&.activitypub?
handle_activitypub
else
handle_ostatus
end
next if (@account.present? && !@account.activitypub?) || actor_json.nil?
@account = ActivityPub::ProcessAccountService.new.call(@username, @domain, actor_json)
else
raise Mastodon::RaceConditionError
end
end
@account
rescue Goldfinger::Error => e
Rails.logger.debug "Webfinger query for #{uri} unsuccessful: #{e}"
nil
end
private
def links_missing?
!(activitypub_ready? || ostatus_ready?)
end
def ostatus_ready?
!(@webfinger.link('http://schemas.google.com/g/2010#updates-from').nil? ||
@webfinger.link('salmon').nil? ||
@webfinger.link('http://webfinger.net/rel/profile-page').nil? ||
@webfinger.link('magic-public-key').nil? ||
canonical_uri.nil? ||
hub_url.nil?)
end
def webfinger_update_due?
@ -91,113 +115,13 @@ class ResolveAccountService < BaseService
end
def activitypub_ready?
!@webfinger.link('self').nil? &&
['application/activity+json', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'].include?(@webfinger.link('self').type) &&
!actor_json.nil? &&
actor_json['inbox'].present?
end
def handle_ostatus
create_account if @account.nil?
update_account
update_account_profile if update_profile?
end
def update_profile?
@options[:update_profile]
end
def handle_activitypub
return if actor_json.nil?
@account = ActivityPub::ProcessAccountService.new.call(@username, @domain, actor_json)
rescue Oj::ParseError
nil
end
def create_account
Rails.logger.debug "Creating new remote account for #{@username}@#{@domain}"
@account = Account.new(username: @username, domain: @domain)
@account.suspended_at = domain_block.created_at if auto_suspend?
@account.silenced_at = domain_block.created_at if auto_silence?
@account.private_key = nil
end
def update_account
@account.last_webfingered_at = Time.now.utc
@account.protocol = :ostatus
@account.remote_url = atom_url
@account.salmon_url = salmon_url
@account.url = url
@account.public_key = public_key
@account.uri = canonical_uri
@account.hub_url = hub_url
@account.save!
end
def auto_suspend?
domain_block&.suspend?
end
def auto_silence?
domain_block&.silence?
end
def domain_block
return @domain_block if defined?(@domain_block)
@domain_block = DomainBlock.rule_for(@domain)
end
def atom_url
@atom_url ||= @webfinger.link('http://schemas.google.com/g/2010#updates-from').href
end
def salmon_url
@salmon_url ||= @webfinger.link('salmon').href
!@webfinger.link('self').nil? && ['application/activity+json', 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'].include?(@webfinger.link('self').type)
end
def actor_url
@actor_url ||= @webfinger.link('self').href
end
def url
@url ||= @webfinger.link('http://webfinger.net/rel/profile-page').href
end
def public_key
@public_key ||= magic_key_to_pem(@webfinger.link('magic-public-key').href)
end
def canonical_uri
return @canonical_uri if defined?(@canonical_uri)
author_uri = atom.at_xpath('/xmlns:feed/xmlns:author/xmlns:uri')
if author_uri.nil?
owner = atom.at_xpath('/xmlns:feed').at_xpath('./dfrn:owner', dfrn: DFRN_NS)
author_uri = owner.at_xpath('./xmlns:uri') unless owner.nil?
end
@canonical_uri = author_uri.nil? ? nil : author_uri.content
end
def hub_url
return @hub_url if defined?(@hub_url)
hubs = atom.xpath('//xmlns:link[@rel="hub"]')
@hub_url = hubs.empty? || hubs.first['href'].nil? ? nil : hubs.first['href']
end
def atom_body
return @atom_body if defined?(@atom_body)
@atom_body = Request.new(:get, atom_url).perform do |response|
raise Mastodon::UnexpectedResponseError, response unless response.code == 200
response.body_with_limit
end
end
def actor_json
return @actor_json if defined?(@actor_json)
@ -205,15 +129,6 @@ class ResolveAccountService < BaseService
@actor_json = supported_context?(json) && equals_or_includes_any?(json['type'], ActivityPub::FetchRemoteAccountService::SUPPORTED_TYPES) ? json : nil
end
def atom
return @atom if defined?(@atom)
@atom = Nokogiri::XML(atom_body)
end
def update_account_profile
RemoteProfileUpdateWorker.perform_async(@account.id, atom_body.force_encoding('UTF-8'), false)
end
def lock_options
{ redis: Redis.current, key: "resolve:#{@username}@#{@domain}" }
end

58
app/services/resolve_url_service.rb
View File

@ -4,64 +4,51 @@ class ResolveURLService < BaseService
include JsonLdHelper
include Authorization
attr_reader :url
def call(url, on_behalf_of: nil)
@url = url
@url = url
@on_behalf_of = on_behalf_of
return process_local_url if local_url?
process_url unless fetched_atom_feed.nil?
if local_url?
process_local_url
elsif !fetched_resource.nil?
process_url
end
end
private
def process_url
if equals_or_includes_any?(type, %w(Application Group Organization Person Service))
FetchRemoteAccountService.new.call(atom_url, body, protocol)
elsif equals_or_includes_any?(type, %w(Note Article Image Video Page Question))
FetchRemoteStatusService.new.call(atom_url, body, protocol)
if equals_or_includes_any?(type, ActivityPub::FetchRemoteAccountService::SUPPORTED_TYPES)
FetchRemoteAccountService.new.call(resource_url, body, protocol)
elsif equals_or_includes_any?(type, ActivityPub::Activity::Create::SUPPORTED_TYPES + ActivityPub::Activity::Create::CONVERTED_TYPES)
status = FetchRemoteStatusService.new.call(resource_url, body, protocol)
authorize_with @on_behalf_of, status, :show? unless status.nil?
status
end
end
def fetched_atom_feed
@_fetched_atom_feed ||= FetchAtomService.new.call(url)
def fetched_resource
@fetched_resource ||= FetchResourceService.new.call(@url)
end
def atom_url
fetched_atom_feed.first
def resource_url
fetched_resource.first
end
def body
fetched_atom_feed.second[:prefetched_body]
fetched_resource.second[:prefetched_body]
end
def protocol
fetched_atom_feed.third
fetched_resource.third
end
def type
return json_data['type'] if protocol == :activitypub
case xml_root
when 'feed'
'Person'
when 'entry'
'Note'
end
end
def json_data
@_json_data ||= body_to_json(body)
end
def xml_root
xml_data.root.name
end
def xml_data
@_xml_data ||= Nokogiri::XML(body, nil, 'utf-8')
@json_data ||= body_to_json(body)
end
def local_url?
@ -73,10 +60,7 @@ class ResolveURLService < BaseService
return unless recognized_params[:action] == 'show'
if recognized_params[:controller] == 'stream_entries'
status = StreamEntry.find_by(id: recognized_params[:id])&.status
check_local_status(status)
elsif recognized_params[:controller] == 'statuses'
if recognized_params[:controller] == 'statuses'
status = Status.find_by(id: recognized_params[:id])
check_local_status(status)
elsif recognized_params[:controller] == 'accounts'
@ -86,10 +70,10 @@ class ResolveURLService < BaseService
def check_local_status(status)
return if status.nil?
authorize_with @on_behalf_of, status, :show?
status
rescue Mastodon::NotPermittedError
# Do not disclose the existence of status the user is not authorized to see
nil
end
end

22
app/services/search_service.rb
View File

@ -11,7 +11,7 @@ class SearchService < BaseService
default_results.tap do |results|
if url_query?
results.merge!(url_resource_results) unless url_resource.nil?
results.merge!(url_resource_results) unless url_resource.nil? || (@options[:type].present? && url_resource_symbol != @options[:type].to_sym)
elsif @query.present?
results[:accounts] = perform_accounts_search! if account_searchable?
results[:statuses] = perform_statuses_search! if full_text_searchable?
@ -33,8 +33,7 @@ class SearchService < BaseService
end
def perform_statuses_search!
definition = StatusesIndex.filter(term: { searchable_by: @account.id })
.query(multi_match: { type: 'most_fields', query: @query, operator: 'and', fields: %w(text text.stemmed) })
definition = parsed_query.apply(StatusesIndex.filter(term: { searchable_by: @account.id }))
if @options[:account_id].present?
definition = definition.filter(term: { account_id: @options[:account_id] })
@ -53,15 +52,16 @@ class SearchService < BaseService
preloaded_relations = relations_map_for_account(@account, account_ids, account_domains)
results.reject { |status| StatusFilter.new(status, @account, preloaded_relations).filtered? }
rescue Faraday::ConnectionFailed
rescue Faraday::ConnectionFailed, Parslet::ParseFailed
[]
end
def perform_hashtags_search!
Tag.search_for(
@query.gsub(/\A#/, ''),
@limit,
@offset
TagSearchService.new.call(
@query,
limit: @limit,
offset: @offset,
exclude_unreviewed: @options[:exclude_unreviewed]
)
end
@ -70,7 +70,7 @@ class SearchService < BaseService
end
def url_query?
@options[:type].blank? && @query =~ /\Ahttps?:\/\//
@resolve && @query =~ /\Ahttps?:\/\//
end
def url_resource_results
@ -120,4 +120,8 @@ class SearchService < BaseService
domain_blocking_by_domain: Account.domain_blocking_map_by_domain(domains, account.id),
}
end
def parsed_query
SearchQueryTransformer.new.apply(SearchQueryParser.new.parse(@query))
end
end

39
app/services/send_interaction_service.rb
View File

@ -1,39 +0,0 @@
# frozen_string_literal: true
class SendInteractionService < BaseService
# Send an Atom representation of an interaction to a remote Salmon endpoint
# @param [String] Entry XML
# @param [Account] source_account
# @param [Account] target_account
def call(xml, source_account, target_account)
@xml = xml
@source_account = source_account
@target_account = target_account
return if !target_account.ostatus? || block_notification?
build_request.perform do |delivery|
raise Mastodon::UnexpectedResponseError, delivery unless delivery.code > 199 && delivery.code < 300
end
end
private
def build_request
request = Request.new(:post, @target_account.salmon_url, body: envelope)
request.add_headers('Content-Type' => 'application/magic-envelope+xml')
request
end
def envelope
salmon.pack(@xml, @source_account.keypair)
end
def block_notification?
DomainBlock.blocked?(@target_account.domain)
end
def salmon
@salmon ||= OStatus2::Salmon.new
end
end

58
app/services/subscribe_service.rb
View File

@ -1,58 +0,0 @@
# frozen_string_literal: true
class SubscribeService < BaseService
def call(account)
return if account.hub_url.blank?
@account = account
@account.secret = SecureRandom.hex
build_request.perform do |response|
if response_failed_permanently? response
# We're not allowed to subscribe. Fail and move on.
@account.secret = ''
@account.save!
elsif response_successful? response
# The subscription will be confirmed asynchronously.
@account.save!
else
# The response was either a 429 rate limit, or a 5xx error.
# We need to retry at a later time. Fail loudly!
raise Mastodon::UnexpectedResponseError, response
end
end
end
private
def build_request
request = Request.new(:post, @account.hub_url, form: subscription_params)
request.on_behalf_of(some_local_account) if some_local_account
request
end
def subscription_params
{
'hub.topic': @account.remote_url,
'hub.mode': 'subscribe',
'hub.callback': api_subscription_url(@account.id),
'hub.verify': 'async',
'hub.secret': @account.secret,
'hub.lease_seconds': 7.days.seconds,
}
end
def some_local_account
@some_local_account ||= Account.local.without_suspended.first
end
# Any response in the 3xx or 4xx range, except for 429 (rate limit)
def response_failed_permanently?(response)
(response.status.redirect? || response.status.client_error?) && !response.status.too_many_requests?
end
# Any response in the 2xx range
def response_successful?(response)
response.status.success?
end
end

63
app/services/suspend_account_service.rb
View File

@ -15,7 +15,6 @@ class SuspendAccountService < BaseService
favourites
follow_requests
list_accounts
media_attachments
mute_relationships
muted_by_relationships
notifications
@ -24,8 +23,6 @@ class SuspendAccountService < BaseService
report_notes
scheduled_statuses
status_pins
stream_entries
subscriptions
).freeze
ASSOCIATIONS_ON_DESTROY = %w(
@ -34,14 +31,26 @@ class SuspendAccountService < BaseService
targeted_reports
).freeze
# Suspend an account and remove as much of its data as possible
# Suspend or remove an account and remove as much of its data
# as possible. If it's a local account and it has not been confirmed
# or never been approved, then side effects are skipped and both
# the user and account records are removed fully. Otherwise,
# it is controlled by options.
# @param [Account]
# @param [Hash] options
# @option [Boolean] :including_user Remove the user record as well
# @option [Boolean] :destroy Remove the account record instead of suspending
# @option [Boolean] :reserve_email Keep user record. Only applicable for local accounts
# @option [Boolean] :reserve_username Keep account record
# @option [Boolean] :skip_side_effects Side effects are ActivityPub and streaming API payloads
# @option [Time] :suspended_at Only applicable when :reserve_username is true
def call(account, **options)
@account = account
@options = options
@options = { reserve_username: true, reserve_email: true }.merge(options)
if @account.local? && @account.user_unconfirmed_or_pending?
@options[:reserve_email] = false
@options[:reserve_username] = false
@options[:skip_side_effects] = true
end
reject_follows!
purge_user!
@ -62,26 +71,39 @@ class SuspendAccountService < BaseService
def purge_user!
return if !@account.local? || @account.user.nil?
if @options[:including_user]
@account.user.destroy
else
if @options[:reserve_email]
@account.user.disable!
@account.user.invites.where(uses: 0).destroy_all
else
@account.user.destroy
end
end
def purge_content!
distribute_delete_actor! if @account.local? && !@options[:skip_distribution]
distribute_delete_actor! if @account.local? && !@options[:skip_side_effects]
@account.statuses.reorder(nil).find_in_batches do |statuses|
BatchedRemoveStatusService.new.call(statuses, skip_side_effects: @options[:destroy])
statuses.reject! { |status| reported_status_ids.include?(status.id) } if @options[:reserve_username]
BatchedRemoveStatusService.new.call(statuses, skip_side_effects: @options[:skip_side_effects])
end
@account.media_attachments.reorder(nil).find_each do |media_attachment|
next if @options[:reserve_username] && reported_status_ids.include?(media_attachment.status_id)
media_attachment.destroy
end
@account.polls.reorder(nil).find_each do |poll|
next if @options[:reserve_username] && reported_status_ids.include?(poll.status_id)
poll.destroy
end
associations_for_destruction.each do |association_name|
destroy_all(@account.public_send(association_name))
end
@account.destroy if @options[:destroy]
@account.destroy unless @options[:reserve_username]
end
def purge_profile!
@ -89,11 +111,13 @@ class SuspendAccountService < BaseService
# there is no point wasting time updating
# its values first
return if @options[:destroy]
return unless @options[:reserve_username]
@account.silenced_at = nil
@account.suspended_at = @options[:suspended_at] || Time.now.utc
@account.locked = false
@account.memorial = false
@account.discoverable = false
@account.display_name = ''
@account.note = ''
@account.fields = []
@ -101,6 +125,7 @@ class SuspendAccountService < BaseService
@account.followers_count = 0
@account.following_count = 0
@account.moved_to_account = nil
@account.trust_level = :untrusted
@account.avatar.destroy
@account.header.destroy
@account.save!
@ -136,11 +161,15 @@ class SuspendAccountService < BaseService
Account.inboxes - delivery_inboxes
end
def reported_status_ids
@reported_status_ids ||= Report.where(target_account: @account).unresolved.pluck(:status_ids).flatten.uniq
end
def associations_for_destruction
if @options[:destroy]
ASSOCIATIONS_ON_SUSPEND + ASSOCIATIONS_ON_DESTROY
else
if @options[:reserve_username]
ASSOCIATIONS_ON_SUSPEND
else
ASSOCIATIONS_ON_SUSPEND + ASSOCIATIONS_ON_DESTROY
end
end
end

87
app/services/tag_search_service.rb Normal file
View File

@ -0,0 +1,87 @@
# frozen_string_literal: true
class TagSearchService < BaseService
def call(query, options = {})
@query = query.strip.gsub(/\A#/, '')
@offset = options.delete(:offset).to_i
@limit = options.delete(:limit).to_i
@options = options
results = from_elasticsearch if Chewy.enabled?
results ||= from_database
results
end
private
def from_elasticsearch
query = {
function_score: {
query: {
multi_match: {
query: @query,
fields: %w(name.edge_ngram name),
type: 'most_fields',
operator: 'and',
},
},
functions: [
{
field_value_factor: {
field: 'usage',
modifier: 'log2p',
missing: 0,
},
},
{
gauss: {
last_status_at: {
scale: '7d',
offset: '14d',
decay: 0.5,
},
},
},
],
boost_mode: 'multiply',
},
}
filter = {
bool: {
should: [
{
term: {
reviewed: {
value: true,
},
},
},
{
match: {
name: {
query: @query,
},
},
},
],
},
}
definition = TagsIndex.query(query)
definition = definition.filter(filter) if @options[:exclude_unreviewed]
definition.limit(@limit).offset(@offset).objects.compact
rescue Faraday::ConnectionFailed, Parslet::ParseFailed
nil
end
def from_database
Tag.search_for(@query, @limit, @offset, @options)
end
end

11
app/services/unallow_domain_service.rb Normal file
View File

@ -0,0 +1,11 @@
# frozen_string_literal: true
class UnallowDomainService < BaseService
def call(domain_allow)
Account.where(domain: domain_allow.domain).find_each do |account|
SuspendAccountService.new.call(account, reserve_username: false)
end
domain_allow.destroy
end
end

19
app/services/unblock_domain_service.rb
View File

@ -10,24 +10,9 @@ class UnblockDomainService < BaseService
end
def process_retroactive_updates
blocked_accounts.in_batches.update_all(update_options) unless domain_block.noop?
end
def blocked_accounts
scope = Account.by_domain_and_subdomains(domain_block.domain)
if domain_block.silence?
scope.where(silenced_at: @domain_block.created_at)
else
scope.where(suspended_at: @domain_block.created_at)
end
end
def update_options
{ domain_block_impact => nil }
end
def domain_block_impact
domain_block.silence? ? :silenced_at : :suspended_at
scope.where(silenced_at: domain_block.created_at).in_batches.update_all(silenced_at: nil) unless domain_block.noop?
scope.where(suspended_at: domain_block.created_at).in_batches.update_all(suspended_at: nil) if domain_block.suspend?
end
end

12
app/services/unblock_service.rb
View File

@ -7,25 +7,17 @@ class UnblockService < BaseService
return unless account.blocking?(target_account)
unblock = account.unblock!(target_account)
create_notification(unblock) unless target_account.local?
create_notification(unblock) if !target_account.local? && target_account.activitypub?
unblock
end
private
def create_notification(unblock)
if unblock.target_account.ostatus?
NotificationWorker.perform_async(build_xml(unblock), unblock.account_id, unblock.target_account_id)
elsif unblock.target_account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(unblock), unblock.account_id, unblock.target_account.inbox_url)
end
ActivityPub::DeliveryWorker.perform_async(build_json(unblock), unblock.account_id, unblock.target_account.inbox_url)
end
def build_json(unblock)
Oj.dump(serialize_payload(unblock, ActivityPub::UndoBlockSerializer))
end
def build_xml(block)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.unblock_salmon(block))
end
end

13
app/services/unfavourite_service.rb
View File

@ -6,7 +6,7 @@ class UnfavouriteService < BaseService
def call(account, status)
favourite = Favourite.find_by!(account: account, status: status)
favourite.destroy!
create_notification(favourite) unless status.local?
create_notification(favourite) if !status.account.local? && status.account.activitypub?
favourite
end
@ -14,19 +14,10 @@ class UnfavouriteService < BaseService
def create_notification(favourite)
status = favourite.status
if status.account.ostatus?
NotificationWorker.perform_async(build_xml(favourite), favourite.account_id, status.account_id)
elsif status.account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(favourite), favourite.account_id, status.account.inbox_url)
end
ActivityPub::DeliveryWorker.perform_async(build_json(favourite), favourite.account_id, status.account.inbox_url)
end
def build_json(favourite)
Oj.dump(serialize_payload(favourite, ActivityPub::UndoLikeSerializer))
end
def build_xml(favourite)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.unfavourite_salmon(favourite))
end
end

27
app/services/unfollow_service.rb
View File

@ -6,9 +6,12 @@ class UnfollowService < BaseService
# Unfollow and notify the remote user
# @param [Account] source_account Where to unfollow from
# @param [Account] target_account Which to unfollow
def call(source_account, target_account)
# @param [Hash] options
# @option [Boolean] :skip_unmerge
def call(source_account, target_account, options = {})
@source_account = source_account
@target_account = target_account
@options = options
unfollow! || undo_follow_request!
end
@ -21,9 +24,11 @@ class UnfollowService < BaseService
return unless follow
follow.destroy!
create_notification(follow) unless @target_account.local?
create_reject_notification(follow) if @target_account.local? && !@source_account.local?
UnmergeWorker.perform_async(@target_account.id, @source_account.id)
create_notification(follow) if !@target_account.local? && @target_account.activitypub?
create_reject_notification(follow) if @target_account.local? && !@source_account.local? && @source_account.activitypub?
UnmergeWorker.perform_async(@target_account.id, @source_account.id) unless @options[:skip_unmerge]
follow
end
@ -33,21 +38,17 @@ class UnfollowService < BaseService
return unless follow_request
follow_request.destroy!
create_notification(follow_request) unless @target_account.local?
follow_request
end
def create_notification(follow)
if follow.target_account.ostatus?
NotificationWorker.perform_async(build_xml(follow), follow.account_id, follow.target_account_id)
elsif follow.target_account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_json(follow), follow.account_id, follow.target_account.inbox_url)
end
ActivityPub::DeliveryWorker.perform_async(build_json(follow), follow.account_id, follow.target_account.inbox_url)
end
def create_reject_notification(follow)
# Rejecting an already-existing follow request
return unless follow.account.activitypub?
ActivityPub::DeliveryWorker.perform_async(build_reject_json(follow), follow.target_account_id, follow.account.inbox_url)
end
@ -58,8 +59,4 @@ class UnfollowService < BaseService
def build_reject_json(follow)
Oj.dump(serialize_payload(follow, ActivityPub::RejectFollowSerializer))
end
def build_xml(follow)
OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.unfollow_salmon(follow))
end
end

36
app/services/unsubscribe_service.rb
View File

@ -1,36 +0,0 @@
# frozen_string_literal: true
class UnsubscribeService < BaseService
def call(account)
return if account.hub_url.blank?
@account = account
begin
build_request.perform do |response|
Rails.logger.debug "PuSH unsubscribe for #{@account.acct} failed: #{response.status}" unless response.status.success?
end
rescue HTTP::Error, OpenSSL::SSL::SSLError => e
Rails.logger.debug "PuSH unsubscribe for #{@account.acct} failed: #{e}"
end
@account.secret = ''
@account.subscription_expires_at = nil
@account.save!
end
private
def build_request
Request.new(:post, @account.hub_url, form: subscription_params)
end
def subscription_params
{
'hub.topic': @account.remote_url,
'hub.mode': 'unsubscribe',
'hub.callback': api_subscription_url(@account.id),
'hub.verify': 'async',
}
end
end

4
app/services/update_account_service.rb
View File

@ -20,7 +20,9 @@ class UpdateAccountService < BaseService
private
def authorize_all_follow_requests(account)
AuthorizeFollowWorker.push_bulk(FollowRequest.where(target_account: account).select(:account_id, :target_account_id)) do |req|
follow_requests = FollowRequest.where(target_account: account)
follow_requests = follow_requests.preload(:account).select { |req| !req.account.silenced? }
AuthorizeFollowWorker.push_bulk(follow_requests) do |req|
[req.account_id, req.target_account_id]
end
end

66
app/services/update_remote_profile_service.rb
View File

@ -1,66 +0,0 @@
# frozen_string_literal: true
class UpdateRemoteProfileService < BaseService
attr_reader :account, :remote_profile
def call(body, account, resubscribe = false)
@account = account
@remote_profile = RemoteProfile.new(body)
return if remote_profile.root.nil?
update_account unless remote_profile.author.nil?
old_hub_url = account.hub_url
account.hub_url = remote_profile.hub_link if remote_profile.hub_link.present? && remote_profile.hub_link != old_hub_url
account.save_with_optional_media!
Pubsubhubbub::SubscribeWorker.perform_async(account.id) if resubscribe && account.hub_url != old_hub_url
end
private
def update_account
account.display_name = remote_profile.display_name || ''
account.note = remote_profile.note || ''
account.locked = remote_profile.locked?
if !account.suspended? && !DomainBlock.reject_media?(account.domain)
if remote_profile.avatar.present?
account.avatar_remote_url = remote_profile.avatar
else
account.avatar_remote_url = ''
account.avatar.destroy
end
if remote_profile.header.present?
account.header_remote_url = remote_profile.header
else
account.header_remote_url = ''
account.header.destroy
end
save_emojis if remote_profile.emojis.present?
end
end
def save_emojis
do_not_download = DomainBlock.reject_media?(account.domain)
return if do_not_download
remote_profile.emojis.each do |link|
next unless link['href'] && link['name']
shortcode = link['name'].delete(':')
emoji = CustomEmoji.find_by(shortcode: shortcode, domain: account.domain)
next unless emoji.nil?
emoji = CustomEmoji.new(shortcode: shortcode, domain: account.domain)
emoji.image_remote_url = link['href']
emoji.save
end
end
end

26
app/services/verify_salmon_service.rb
View File

@ -1,26 +0,0 @@
# frozen_string_literal: true
class VerifySalmonService < BaseService
include AuthorExtractor
def call(payload)
body = salmon.unpack(payload)
xml = Nokogiri::XML(body)
xml.encoding = 'utf-8'
account = author_from_xml(xml.at_xpath('/xmlns:entry', xmlns: OStatus::TagManager::XMLNS))
if account.nil?
false
else
salmon.verify(payload, account.keypair)
end
end
private
def salmon
@salmon ||= OStatus2::Salmon.new
end
end

32
app/services/vote_service.rb
View File

@ -12,12 +12,24 @@ class VoteService < BaseService
@choices = choices
@votes = []
ApplicationRecord.transaction do
@choices.each do |choice|
@votes << @poll.votes.create!(account: @account, choice: choice)
already_voted = true
RedisLock.acquire(lock_options) do |lock|
if lock.acquired?
already_voted = @poll.votes.where(account: @account).exists?
ApplicationRecord.transaction do
@choices.each do |choice|
@votes << @poll.votes.create!(account: @account, choice: choice)
end
end
else
raise Mastodon::RaceConditionError
end
end
increment_voters_count! unless already_voted
ActivityTracker.increment('activity:interactions')
if @poll.account.local?
@ -53,4 +65,18 @@ class VoteService < BaseService
def build_json(vote)
Oj.dump(serialize_payload(vote, ActivityPub::VoteSerializer))
end
def increment_voters_count!
unless @poll.voters_count.nil?
@poll.voters_count = @poll.voters_count + 1
@poll.save
end
rescue ActiveRecord::StaleObjectError
@poll.reload
retry
end
def lock_options
{ redis: Redis.current, key: "vote:#{@poll.id}:#{@account.id}" }
end
end