ActivityPub delivery (#4566)
* Deliver ActivityPub Like * Deliver ActivityPub Undo-Like * Deliver ActivityPub Create/Announce activities * Deliver ActivityPub creates from mentions * Deliver ActivityPub Block/Undo-Block * Deliver ActivityPub Accept/Reject-Follow * Deliver ActivityPub Undo-Follow * Deliver ActivityPub Follow * Deliver ActivityPub Delete activities Incidentally fix #889 * Adjust BatchedRemoveStatusService for ActivityPub * Add tests for ActivityPub workers * Add tests for FollowService * Add tests for FavouriteService, UnfollowService and PostStatusService * Add tests for ReblogService, BlockService, UnblockService, ProcessMentionsService * Add tests for AuthorizeFollowService, RejectFollowService, RemoveStatusService * Add tests for BatchedRemoveStatusService * Deliver updates to a local account to ActivityPub followers * Minor adjustmentsth-downstream
Eugen Rochko
7 years ago
committed by
GitHub
parent
0e2a3049e7
commit
5516767c75
@ -0,0 +1,37 @@
|
|||||||
|
# frozen_string_literal: true
|
||||||
|
|
||||||
|
class ActivityPub::DeliveryWorker
|
||||||
|
include Sidekiq::Worker
|
||||||
|
|
||||||
|
sidekiq_options queue: 'push', retry: 5, dead: false
|
||||||
|
|
||||||
|
HEADERS = { 'Content-Type' => 'application/activity+json' }.freeze
|
||||||
|
|
||||||
|
def perform(json, source_account_id, inbox_url)
|
||||||
|
@json = json
|
||||||
|
@source_account = Account.find(source_account_id)
|
||||||
|
@inbox_url = inbox_url
|
||||||
|
|
||||||
|
perform_request
|
||||||
|
|
||||||
|
raise Mastodon::UnexpectedResponseError, @response unless response_successful?
|
||||||
|
rescue => e
|
||||||
|
raise e.class, "Delivery failed for #{inbox_url}: #{e.message}"
|
||||||
|
end
|
||||||
|
|
||||||
|
private
|
||||||
|
|
||||||
|
def build_request
|
||||||
|
request = Request.new(:post, @inbox_url, body: @json)
|
||||||
|
request.on_behalf_of(@source_account, :uri)
|
||||||
|
request.add_headers(HEADERS)
|
||||||
|
end
|
||||||
|
|
||||||
|
def perform_request
|
||||||
|
@response = build_request.perform
|
||||||
|
end
|
||||||
|
|
||||||
|
def response_successful?
|
||||||
|
@response.code > 199 && @response.code < 300
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -0,0 +1,38 @@
|
|||||||
|
# frozen_string_literal: true
|
||||||
|
|
||||||
|
class ActivityPub::DistributionWorker
|
||||||
|
include Sidekiq::Worker
|
||||||
|
|
||||||
|
sidekiq_options queue: 'push'
|
||||||
|
|
||||||
|
def perform(status_id)
|
||||||
|
@status = Status.find(status_id)
|
||||||
|
@account = @status.account
|
||||||
|
|
||||||
|
return if skip_distribution?
|
||||||
|
|
||||||
|
ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
|
||||||
|
[payload, @account.id, inbox_url]
|
||||||
|
end
|
||||||
|
rescue ActiveRecord::RecordNotFound
|
||||||
|
true
|
||||||
|
end
|
||||||
|
|
||||||
|
private
|
||||||
|
|
||||||
|
def skip_distribution?
|
||||||
|
@status.direct_visibility?
|
||||||
|
end
|
||||||
|
|
||||||
|
def inboxes
|
||||||
|
@inboxes ||= @account.followers.inboxes
|
||||||
|
end
|
||||||
|
|
||||||
|
def payload
|
||||||
|
@payload ||= ActiveModelSerializers::SerializableResource.new(
|
||||||
|
@status,
|
||||||
|
serializer: ActivityPub::ActivitySerializer,
|
||||||
|
adapter: ActivityPub::Adapter
|
||||||
|
).to_json
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -0,0 +1,31 @@
|
|||||||
|
# frozen_string_literal: true
|
||||||
|
|
||||||
|
class ActivityPub::UpdateDistributionWorker
|
||||||
|
include Sidekiq::Worker
|
||||||
|
|
||||||
|
sidekiq_options queue: 'push'
|
||||||
|
|
||||||
|
def perform(account_id)
|
||||||
|
@account = Account.find(account_id)
|
||||||
|
|
||||||
|
ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
|
||||||
|
[payload, @account.id, inbox_url]
|
||||||
|
end
|
||||||
|
rescue ActiveRecord::RecordNotFound
|
||||||
|
true
|
||||||
|
end
|
||||||
|
|
||||||
|
private
|
||||||
|
|
||||||
|
def inboxes
|
||||||
|
@inboxes ||= @account.followers.inboxes
|
||||||
|
end
|
||||||
|
|
||||||
|
def payload
|
||||||
|
@payload ||= ActiveModelSerializers::SerializableResource.new(
|
||||||
|
@account,
|
||||||
|
serializer: ActivityPub::UpdateSerializer,
|
||||||
|
adapter: ActivityPub::Adapter
|
||||||
|
).to_json
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -1,22 +1,44 @@
|
|||||||
require 'rails_helper'
|
require 'rails_helper'
|
||||||
|
|
||||||
RSpec.describe ProcessMentionsService do
|
RSpec.describe ProcessMentionsService do
|
||||||
let(:account) { Fabricate(:account, username: 'alice') }
|
let(:account) { Fabricate(:account, username: 'alice') }
|
||||||
let(:remote_user) { Fabricate(:account, username: 'remote_user', domain: 'example.com', salmon_url: 'http://salmon.example.com') }
|
let(:status) { Fabricate(:status, account: account, text: "Hello @#{remote_user.acct}") }
|
||||||
let(:status) { Fabricate(:status, account: account, text: "Hello @#{remote_user.acct}") }
|
|
||||||
|
|
||||||
subject { ProcessMentionsService.new }
|
context 'OStatus' do
|
||||||
|
let(:remote_user) { Fabricate(:account, username: 'remote_user', protocol: :ostatus, domain: 'example.com', salmon_url: 'http://salmon.example.com') }
|
||||||
|
|
||||||
before do
|
subject { ProcessMentionsService.new }
|
||||||
stub_request(:post, remote_user.salmon_url)
|
|
||||||
subject.(status)
|
before do
|
||||||
end
|
stub_request(:post, remote_user.salmon_url)
|
||||||
|
subject.call(status)
|
||||||
|
end
|
||||||
|
|
||||||
it 'creates a mention' do
|
it 'creates a mention' do
|
||||||
expect(remote_user.mentions.where(status: status).count).to eq 1
|
expect(remote_user.mentions.where(status: status).count).to eq 1
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'posts to remote user\'s Salmon end point' do
|
||||||
|
expect(a_request(:post, remote_user.salmon_url)).to have_been_made.once
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'posts to remote user\'s Salmon end point' do
|
context 'ActivityPub' do
|
||||||
expect(a_request(:post, remote_user.salmon_url)).to have_been_made
|
let(:remote_user) { Fabricate(:account, username: 'remote_user', protocol: :activitypub, domain: 'example.com', inbox_url: 'http://example.com/inbox') }
|
||||||
|
|
||||||
|
subject { ProcessMentionsService.new }
|
||||||
|
|
||||||
|
before do
|
||||||
|
stub_request(:post, remote_user.inbox_url)
|
||||||
|
subject.call(status)
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'creates a mention' do
|
||||||
|
expect(remote_user.mentions.where(status: status).count).to eq 1
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'sends activity to the inbox' do
|
||||||
|
expect(a_request(:post, remote_user.inbox_url)).to have_been_made.once
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|||||||
@ -0,0 +1,23 @@
|
|||||||
|
# frozen_string_literal: true
|
||||||
|
|
||||||
|
require 'rails_helper'
|
||||||
|
|
||||||
|
describe ActivityPub::DeliveryWorker do
|
||||||
|
subject { described_class.new }
|
||||||
|
|
||||||
|
let(:sender) { Fabricate(:account) }
|
||||||
|
let(:payload) { 'test' }
|
||||||
|
|
||||||
|
describe 'perform' do
|
||||||
|
it 'performs a request' do
|
||||||
|
stub_request(:post, 'https://example.com/api').to_return(status: 200)
|
||||||
|
subject.perform(payload, sender.id, 'https://example.com/api')
|
||||||
|
expect(a_request(:post, 'https://example.com/api')).to have_been_made.once
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'raises when request fails' do
|
||||||
|
stub_request(:post, 'https://example.com/api').to_return(status: 500)
|
||||||
|
expect { subject.perform(payload, sender.id, 'https://example.com/api') }.to raise_error Mastodon::UnexpectedResponseError
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -0,0 +1,48 @@
|
|||||||
|
require 'rails_helper'
|
||||||
|
|
||||||
|
describe ActivityPub::DistributionWorker do
|
||||||
|
subject { described_class.new }
|
||||||
|
|
||||||
|
let(:status) { Fabricate(:status) }
|
||||||
|
let(:follower) { Fabricate(:account, protocol: :activitypub, inbox_url: 'http://example.com') }
|
||||||
|
|
||||||
|
describe '#perform' do
|
||||||
|
before do
|
||||||
|
allow(ActivityPub::DeliveryWorker).to receive(:push_bulk)
|
||||||
|
follower.follow!(status.account)
|
||||||
|
end
|
||||||
|
|
||||||
|
context 'with public status' do
|
||||||
|
before do
|
||||||
|
status.update(visibility: :public)
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'delivers to followers' do
|
||||||
|
subject.perform(status.id)
|
||||||
|
expect(ActivityPub::DeliveryWorker).to have_received(:push_bulk).with(['http://example.com'])
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
context 'with private status' do
|
||||||
|
before do
|
||||||
|
status.update(visibility: :private)
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'delivers to followers' do
|
||||||
|
subject.perform(status.id)
|
||||||
|
expect(ActivityPub::DeliveryWorker).to have_received(:push_bulk).with(['http://example.com'])
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
context 'with direct status' do
|
||||||
|
before do
|
||||||
|
status.update(visibility: :direct)
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'does nothing' do
|
||||||
|
subject.perform(status.id)
|
||||||
|
expect(ActivityPub::DeliveryWorker).to_not have_received(:push_bulk)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -0,0 +1,15 @@
|
|||||||
|
require 'rails_helper'
|
||||||
|
|
||||||
|
describe ActivityPub::ProcessingWorker do
|
||||||
|
subject { described_class.new }
|
||||||
|
|
||||||
|
let(:account) { Fabricate(:account) }
|
||||||
|
|
||||||
|
describe '#perform' do
|
||||||
|
it 'delegates to ActivityPub::ProcessCollectionService' do
|
||||||
|
allow(ActivityPub::ProcessCollectionService).to receive(:new).and_return(double(:service, call: nil))
|
||||||
|
subject.perform(account.id, '')
|
||||||
|
expect(ActivityPub::ProcessCollectionService).to have_received(:new)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -0,0 +1,16 @@
|
|||||||
|
require 'rails_helper'
|
||||||
|
|
||||||
|
describe ActivityPub::ThreadResolveWorker do
|
||||||
|
subject { described_class.new }
|
||||||
|
|
||||||
|
let(:status) { Fabricate(:status) }
|
||||||
|
let(:parent) { Fabricate(:status) }
|
||||||
|
|
||||||
|
describe '#perform' do
|
||||||
|
it 'gets parent from ActivityPub::FetchRemoteStatusService and glues them together' do
|
||||||
|
allow(ActivityPub::FetchRemoteStatusService).to receive(:new).and_return(double(:service, call: parent))
|
||||||
|
subject.perform(status.id, 'http://example.com/123')
|
||||||
|
expect(status.reload.in_reply_to_id).to eq parent.id
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
@ -0,0 +1,20 @@
|
|||||||
|
require 'rails_helper'
|
||||||
|
|
||||||
|
describe ActivityPub::UpdateDistributionWorker do
|
||||||
|
subject { described_class.new }
|
||||||
|
|
||||||
|
let(:account) { Fabricate(:account) }
|
||||||
|
let(:follower) { Fabricate(:account, protocol: :activitypub, inbox_url: 'http://example.com') }
|
||||||
|
|
||||||
|
describe '#perform' do
|
||||||
|
before do
|
||||||
|
allow(ActivityPub::DeliveryWorker).to receive(:push_bulk)
|
||||||
|
follower.follow!(account)
|
||||||
|
end
|
||||||
|
|
||||||
|
it 'delivers to followers' do
|
||||||
|
subject.perform(account.id)
|
||||||
|
expect(ActivityPub::DeliveryWorker).to have_received(:push_bulk).with(['http://example.com'])
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
Loading…
Reference in new issue