From 4bec613897d8835bb78202c7a36691b654f14967 Mon Sep 17 00:00:00 2001
From: Eugen Rochko <eugen@zeonfederated.com>
Date: Wed, 21 Sep 2016 01:34:14 +0200
Subject: [PATCH] Fix #24 - Thread resolving for remote statuses

This is a big one, so let me enumerate:

Accounts as well as stream entry pages now contain Link headers that
reference the Atom feed and Webfinger URL for the former and Atom entry
for the latter. So you only need to HEAD those resources to get that
information, no need to download and parse HTML <link>s.

ProcessFeedService will now queue ThreadResolveWorker for each remote
status that it cannot find otherwise. Furthermore, entries are now
processed in reverse order (from bottom to top) in case a newer entry
references a chronologically previous one.

ThreadResolveWorker uses FetchRemoteStatusService to obtain a status
and attach the child status it was queued for to it.

FetchRemoteStatusService looks up the URL, first with a HEAD, tests
if it's an Atom feed, in which case it processes it directly. Next
for Link headers to the Atom feed, in which case that is fetched
and processed. Lastly if it's HTML, it is checked for <link>s to the Atom
feed, and if such is found, that is fetched and processed. The account for
the status is derived from author/name attribute in the XML and the hostname
in the URL (domain). FollowRemoteAccountService and ProcessFeedService
are used.

This means that potentially threads are resolved recursively until a dead-end
is encountered, however it is performed asynchronously over background jobs,
so it should be ok.
---
 Gemfile                                       |  1 +
 Gemfile.lock                                  |  2 +
 app/controllers/accounts_controller.rb        |  9 ++-
 app/controllers/stream_entries_controller.rb  |  7 ++
 app/services/fetch_remote_status_service.rb   | 71 +++++++++++++++++++
 app/services/follow_remote_account_service.rb |  2 +-
 app/services/process_feed_service.rb          | 22 +++++-
 app/workers/thread_resolve_worker.rb          | 13 ++++
 .../api/subscriptions_controller_spec.rb      |  8 +++
 9 files changed, 130 insertions(+), 5 deletions(-)
 create mode 100644 app/services/fetch_remote_status_service.rb
 create mode 100644 app/workers/thread_resolve_worker.rb

diff --git a/Gemfile b/Gemfile
index 2e3d9b814cf845..7b8b9f0f34ed10 100644
--- a/Gemfile
+++ b/Gemfile
@@ -21,6 +21,7 @@ gem 'paperclip-av-transcoder'
 gem 'http'
 gem 'addressable'
 gem 'nokogiri'
+gem 'link_header'
 gem 'ostatus2'
 gem 'goldfinger'
 gem 'devise'
diff --git a/Gemfile.lock b/Gemfile.lock
index 6a567adec00133..485a4687a22210 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -149,6 +149,7 @@ GEM
     letter_opener (1.4.1)
       launchy (~> 2.2)
     libv8 (3.16.14.15)
+    link_header (0.0.8)
     lograge (0.4.1)
       actionpack (>= 4, < 5.1)
       activesupport (>= 4, < 5.1)
@@ -368,6 +369,7 @@ DEPENDENCIES
   jbuilder (~> 2.0)
   jquery-rails
   letter_opener
+  link_header
   lograge
   nokogiri
   oj
diff --git a/app/controllers/accounts_controller.rb b/app/controllers/accounts_controller.rb
index 2d4c2dd9a56b2f..bfdc5b6d91ab1a 100644
--- a/app/controllers/accounts_controller.rb
+++ b/app/controllers/accounts_controller.rb
@@ -2,7 +2,7 @@ class AccountsController < ApplicationController
   layout 'public'
 
   before_action :set_account
-  before_action :set_webfinger_header
+  before_action :set_link_headers
 
   def show
     respond_to do |format|
@@ -39,8 +39,11 @@ def set_account
     @account = Account.find_local!(params[:username])
   end
 
-  def set_webfinger_header
-    response.headers['Link'] = "<#{webfinger_account_url}>; rel=\"lrdd\"; type=\"application/xrd+xml\""
+  def set_link_headers
+    response.headers['Link'] = LinkHeader.new([
+      [webfinger_account_url, [['rel', 'lrdd'], ['type', 'application/xrd+xml']]],
+      [account_url(@account, format: 'atom'), [['rel', 'alternate'], ['type', 'application/atom+xml']]]
+    ])
   end
 
   def webfinger_account_url
diff --git a/app/controllers/stream_entries_controller.rb b/app/controllers/stream_entries_controller.rb
index c26149627954a1..e25e7f5edb3d05 100644
--- a/app/controllers/stream_entries_controller.rb
+++ b/app/controllers/stream_entries_controller.rb
@@ -3,6 +3,7 @@ class StreamEntriesController < ApplicationController
 
   before_action :set_account
   before_action :set_stream_entry
+  before_action :set_link_headers
 
   def show
     @type = @stream_entry.activity_type.downcase
@@ -33,6 +34,12 @@ def set_account
     @account = Account.find_local!(params[:account_username])
   end
 
+  def set_link_headers
+    response.headers['Link'] = LinkHeader.new([
+      [account_stream_entry_url(@account, @stream_entry, format: 'atom'), [['rel', 'alternate'], ['type', 'application/atom+xml']]]
+    ])
+  end
+
   def set_stream_entry
     @stream_entry = @account.stream_entries.find(params[:id])
   end
diff --git a/app/services/fetch_remote_status_service.rb b/app/services/fetch_remote_status_service.rb
new file mode 100644
index 00000000000000..c872cb385e646a
--- /dev/null
+++ b/app/services/fetch_remote_status_service.rb
@@ -0,0 +1,71 @@
+class FetchRemoteStatusService < BaseService
+  def call(url)
+    response = http_client.head(url)
+
+    Rails.logger.debug "Remote status HEAD request returned code #{response.code}"
+    return nil if response.code != 200
+
+    if response.mime_type == 'application/atom+xml'
+      return process_atom(url, fetch(url))
+    elsif !response['Link'].blank?
+      return process_headers(response)
+    else
+      return process_html(fetch(url))
+    end
+  end
+
+  private
+
+  def process_atom(url, body)
+    Rails.logger.debug "Processing Atom for remote status"
+
+    xml     = Nokogiri::XML(body)
+    account = extract_author(url, xml)
+
+    return nil if account.nil?
+
+    statuses = ProcessFeedService.new.(body, account)
+
+    return statuses.first
+  end
+
+  def process_html(body)
+    Rails.logger.debug "Processing HTML for remote status"
+
+    page = Nokogiri::HTML(body)
+    alternate_link = page.xpath('//link[@rel="alternate"]').find { |link| link['type'] == 'application/atom+xml' }
+
+    return nil if alternate_link.nil?
+    return process_atom(alternate_link['href'], fetch(alternate_link['href']))
+  end
+
+  def process_headers(response)
+    Rails.logger.debug "Processing link header for remote status"
+
+    link_header    = LinkHeader.parse(response['Link'])
+    alternate_link = link_header.find_link(['rel', 'alternate'], ['type', 'application/atom+xml'])
+
+    return nil if alternate_link.nil?
+    return process_atom(alternate_link.href, fetch(alternate_link.href))
+  end
+
+  def extract_author(url, xml)
+    url_parts = Addressable::URI.parse(url)
+    username  = xml.at_xpath('//xmlns:author/xmlns:name').try(:content)
+    domain    = url_parts.host
+
+    return nil if username.nil?
+
+    Rails.logger.debug "Going to webfinger #{username}@#{domain}"
+
+    return FollowRemoteAccountService.new.("#{username}@#{domain}")
+  end
+
+  def fetch(url)
+    http_client.get(url).to_s
+  end
+
+  def http_client
+    HTTP.timeout(:per_operation, write: 20, connect: 20, read: 50)
+  end
+end
diff --git a/app/services/follow_remote_account_service.rb b/app/services/follow_remote_account_service.rb
index f3a384ecca3a7e..ef3949e3682045 100644
--- a/app/services/follow_remote_account_service.rb
+++ b/app/services/follow_remote_account_service.rb
@@ -72,7 +72,7 @@ def update_remote_profile_service
   end
 
   def http_client
-    HTTP
+    HTTP.timeout(:per_operation, write: 20, connect: 20, read: 50)
   end
 end
 
diff --git a/app/services/process_feed_service.rb b/app/services/process_feed_service.rb
index a692054947a007..49834f5c16c1fe 100644
--- a/app/services/process_feed_service.rb
+++ b/app/services/process_feed_service.rb
@@ -2,10 +2,11 @@ class ProcessFeedService < BaseService
   # Create local statuses from an Atom feed
   # @param [String] body Atom feed
   # @param [Account] account Account this feed belongs to
+  # @return [Enumerable] created statuses
   def call(body, account)
     xml = Nokogiri::XML(body)
     update_remote_profile_service.(xml.at_xpath('/xmlns:feed/xmlns:author'), account) unless xml.at_xpath('/xmlns:feed').nil?
-    xml.xpath('//xmlns:entry').each { |entry| process_entry(account, entry) }
+    xml.xpath('//xmlns:entry').reverse_each.map { |entry| process_entry(account, entry) }.compact
   end
 
   private
@@ -45,6 +46,8 @@ def process_entry(account, entry)
 
       DistributionWorker.perform_async(status.id)
     end
+
+    return status
   end
 
   def record_remote_mentions(status, links)
@@ -103,6 +106,10 @@ def add_reblog!(entry, status)
   def add_reply!(entry, status)
     status.thread = find_original_status(entry, thread_id(entry))
     status.save!
+
+    if status.thread.nil? && !thread_href(entry).nil?
+      ThreadResolveWorker.perform_async(status.id, thread_href(entry))
+    end
   end
 
   def delete_post!(status)
@@ -131,6 +138,13 @@ def fetch_remote_status(xml)
 
     status = Status.new(account: account, uri: target_id(xml), text: target_content(xml), url: target_url(xml), created_at: published(xml), updated_at: updated(xml))
     status.thread = find_original_status(xml, thread_id(xml))
+    status.save
+
+    if status.saved? && status.thread.nil? && !thread_href(xml).nil?
+      ThreadResolveWorker.perform_async(status.id, thread_href(xml))
+    end
+
+    status
   rescue Goldfinger::Error, HTTP::Error
     nil
   end
@@ -153,6 +167,12 @@ def thread_id(xml)
     nil
   end
 
+  def thread_href(xml)
+    xml.at_xpath('./thr:in-reply-to').attribute('href').value
+  rescue
+    nil
+  end
+
   def target_id(xml)
     xml.at_xpath('.//activity:object/xmlns:id').content
   rescue
diff --git a/app/workers/thread_resolve_worker.rb b/app/workers/thread_resolve_worker.rb
new file mode 100644
index 00000000000000..ccc56380b77fa5
--- /dev/null
+++ b/app/workers/thread_resolve_worker.rb
@@ -0,0 +1,13 @@
+class ThreadResolveWorker
+  include Sidekiq::Worker
+
+  def perform(child_status_id, parent_url)
+    child_status  = Status.find(child_status_id)
+    parent_status = FetchRemoteStatusService.new.(parent_url)
+
+    unless parent_status.nil?
+      child_status.thread = parent_status
+      child_status.save!
+    end
+  end
+end
diff --git a/spec/controllers/api/subscriptions_controller_spec.rb b/spec/controllers/api/subscriptions_controller_spec.rb
index ad0d0bc05548ce..e0ae8d48ea3d42 100644
--- a/spec/controllers/api/subscriptions_controller_spec.rb
+++ b/spec/controllers/api/subscriptions_controller_spec.rb
@@ -24,6 +24,14 @@
 
     before do
       stub_request(:get, "https://quitter.no/avatar/7477-300-20160211190340.png").to_return(request_fixture('avatar.txt'))
+      stub_request(:head, "https://quitter.no/notice/1269244").to_return(status: 404)
+      stub_request(:head, "https://quitter.no/notice/1265331").to_return(status: 404)
+      stub_request(:head, "https://community.highlandarrow.com/notice/54411").to_return(status: 404)
+      stub_request(:head, "https://community.highlandarrow.com/notice/53857").to_return(status: 404)
+      stub_request(:head, "https://community.highlandarrow.com/notice/51852").to_return(status: 404)
+      stub_request(:head, "https://social.umeahackerspace.se/notice/424348").to_return(status: 404)
+      stub_request(:head, "https://community.highlandarrow.com/notice/50467").to_return(status: 404)
+      stub_request(:head, "https://quitter.no/notice/1243309").to_return(status: 404)
 
       request.env['HTTP_X_HUB_SIGNATURE'] = "sha1=#{OpenSSL::HMAC.hexdigest('sha1', 'abc', feed)}"
       request.env['RAW_POST_DATA'] = feed