Add Feeds::Import service class (#10998)

* Add the parallel gem

* Add prototype version of Feeds::Import with parallel URL fetching and parsing

* Tune Feeds::Import and add rake task for local tests

* Add rough measurer tool

* Add specs

* Apply suggestions by @citizen428

* Replace silence with silent

* No need for eager loading

* Add temporary Articles::DevFeedsImportWorker

* Remove temporary rake task

* Add basic error handling, copied from RssReader

* Fix error handling

* Remove temporary measuring rake task

* Remove logging added for development purposes

* Add info and error logging on error level

Gemfile

@@ -66,6 +66,7 @@ gem "omniauth", "~> 1.9" # A generalized Rack framework for multiple-provider au
 gem "omniauth-facebook", "~> 8.0" # OmniAuth strategy for Facebook
 gem "omniauth-github", "~> 1.3" # OmniAuth strategy for GitHub
 gem "omniauth-twitter", "~> 1.4" # OmniAuth strategy for Twitter
+gem "parallel", "~> 1.19" # Run any kind of code in parallel processes
 gem "patron", "~> 0.13.3" # HTTP client library based on libcurl, used with Elasticsearch to support http keep-alive connections
 gem "pg", "~> 1.2" # Pg is the Ruby interface to the PostgreSQL RDBMS
 gem "puma", "~> 5.0.2" # Puma is a simple, fast, threaded, and highly concurrent HTTP 1.1 server

Gemfile.lock

@@ -923,6 +923,7 @@ DEPENDENCIES
   omniauth-facebook (~> 8.0)
   omniauth-github (~> 1.3)
   omniauth-twitter (~> 1.4)
+  parallel (~> 1.19)
   patron (~> 0.13.3)
   pg (~> 1.2)
   pry (~> 0.13)

app/services/feeds/import.rb

@@ -0,0 +1,186 @@
+# TODO: [rhymes]
+# => add Feeds::ImportUser to fetch a single user
+# => add Feeds::ValidateFeedUrl to validate a single feed URL
+module Feeds
+  class Import
+    def self.call
+      new.call
+    end
+
+    # TODO: add `users` param
+    def initialize
+      @users = User.where.not(feed_url: [nil, ""])
+
+      # NOTE: should these be configurable? Currently they are the result of empiric
+      # tests trying to find a balance between memory occupation and speed
+      @users_batch_size = 50
+      @num_fetchers = 8
+      @num_parsers = 4
+    end
+
+    def call
+      total_articles_count = 0
+
+      users.in_batches(of: users_batch_size) do |batch_of_users|
+        feeds_per_user_id = fetch_feeds(batch_of_users)
+        Rails.logger.error("feeds::import::feeds_per_user_id.length: #{feeds_per_user_id.length}")
+
+        feedjira_objects = parse_feeds(feeds_per_user_id)
+        Rails.logger.error("feeds::import::feedjira_objects.length: #{feedjira_objects.length}")
+
+        # NOTE: doing this sequentially to avoid locking problems with the DB
+        # and unnecessary conflicts
+        articles = feedjira_objects.flat_map do |user_id, feed|
+          # TODO: replace `feed` with `feed.url` as `RssReader::Assembler`
+          # only actually needs feed.url
+          user = batch_of_users.detect { |u| u.id == user_id }
+          create_articles_from_user_feed(user, feed)
+        end
+        Rails.logger.error("feeds::import::articles.length: #{articles.length}")
+
+        total_articles_count += articles.length
+        Rails.logger.error("feeds::import::total_articles_count: #{total_articles_count}")
+
+        articles.each { |article| Slack::Messengers::ArticleFetchedFeed.call(article: article) }
+      end
+
+      total_articles_count
+    end
+
+    private
+
+    attr_reader :users, :users_batch_size, :num_fetchers, :num_parsers
+
+    # TODO: put this in separate service object
+    def fetch_feeds(batch_of_users)
+      data = batch_of_users.pluck(:id, :feed_url)
+
+      result = Parallel.map(data, in_threads: num_fetchers) do |user_id, url|
+        response = HTTParty.get(url.strip, timeout: 10)
+
+        [user_id, response.body]
+      rescue StandardError => e
+        # TODO: add better exception handling
+        # For example, we should stop pulling feeds that return 404 and disable them?
+
+        report_error(
+          e,
+          feeds_import_info: {
+            user_id: user_id,
+            url: url,
+            error: "Feeds::Import::FetchFeedError"
+          },
+        )
+
+        next
+      end
+
+      batch_of_users.update_all(feed_fetched_at: Time.current)
+
+      result.compact.to_h
+    end
+
+    # TODO: put this in separate service object
+    def parse_feeds(feeds_per_user_id)
+      result = Parallel.map(feeds_per_user_id, in_threads: num_parsers) do |user_id, feed_xml|
+        parsed_feed = Feedjira.parse(feed_xml)
+
+        [user_id, parsed_feed]
+      rescue StandardError => e
+        # TODO: add better exception handling (eg. rescueing Feedjira::NoParserAvailable separately)
+        report_error(
+          e,
+          feeds_import_info: {
+            user_id: user_id,
+            error: "Feeds::Import::ParseFeedError"
+          },
+        )
+
+        next
+      end
+
+      result.compact.to_h
+    end
+
+    # TODO: currently this is exactly as in RSSReader, but we might find
+    # avenues for optimization, like:
+    # 1. why are we sending N exists query to the DB, one per each item, can we fetch them all?
+    # 2. should we queue a batch of workers to create articles, but then, following issues ensue:
+    # => synchronization on write (table/row locking)
+    # => what happens if 2 jobs are in the queue for the same article?
+    # => what happens if they stay in the queue for long and the next iteration of the feeds importer starts?
+    def create_articles_from_user_feed(user, feed)
+      articles = []
+
+      feed.entries.reverse_each do |item|
+        next if medium_reply?(item) || article_exists?(user, item)
+
+        feed_source_url = item.url.strip.split("?source=")[0]
+        article = Article.create!(
+          feed_source_url: feed_source_url,
+          user_id: user.id,
+          published_from_feed: true,
+          show_comments: true,
+          body_markdown: RssReader::Assembler.call(item, user, feed, feed_source_url),
+          organization_id: nil,
+        )
+
+        articles.append(article)
+      rescue StandardError => e
+        # TODO: add better exception handling
+        report_error(
+          e,
+          feeds_import_info: {
+            username: user.username,
+            feed_url: user.feed_url,
+            item_count: get_item_count_error(feed),
+            error: "Feeds::Import::CreateArticleError:#{item.url}"
+          },
+        )
+
+        next
+      end
+
+      articles
+    end
+
+    def get_host_without_www(url)
+      url = "http://#{url}" if URI.parse(url).scheme.nil?
+      host = URI.parse(url).host.downcase
+      host.start_with?("www.") ? host[4..] : host
+    end
+
+    def medium_reply?(item)
+      get_host_without_www(item.url.strip) == "medium.com" &&
+        !item[:categories] &&
+        content_is_not_the_title?(item)
+    end
+
+    def content_is_not_the_title?(item)
+      # [[:space:]] removes all whitespace, including unicode ones.
+      content = item.content.gsub(/[[:space:]]/, " ")
+      title = item.title.delete("…")
+      content.include?(title)
+    end
+
+    def article_exists?(user, item)
+      title = item.title.strip.gsub('"', '\"')
+      feed_source_url = item.url.strip.split("?source=")[0]
+      relation = user.articles
+      relation.where(title: title).or(relation.where(feed_source_url: feed_source_url)).exists?
+    end
+
+    def report_error(error, metadata)
+      Rails.logger.error("feeds::import::error::#{error.class}::#{metadata}")
+      Rails.logger.error(error)
+    end
+
+    def get_item_count_error(feed)
+      if feed
+        feed.entries ? feed.entries.length : "no count"
+      else
+        "NIL FEED, INVALID URL"
+      end
+    end
+  end
+end

app/workers/articles/dev_feeds_import_worker.rb

@@ -0,0 +1,17 @@
+# NOTE: [rhymes]
+# This script will soon be removed. We need it to collect monitoring data on
+# Datadog about the production behavior of the `Feeds::Import` class
+module Articles
+  class DevFeedsImportWorker
+    include Sidekiq::Worker
+
+    sidekiq_options queue: :medium_priority, retry: 10
+
+    def perform
+      return unless SiteConfig.community_name == "DEV"
+      return if Rails.cache.read("cancel_feeds_import").present?
+
+      ::Feeds::Import.call
+    end
+  end
+end

lib/tasks/fetch.rake

@@ -6,6 +6,10 @@ task fetch_all_rss: :environment do
   RssReader.get_all_articles(force: false) # don't force fetch. Fetch "random" subset instead of all of them.
 end
 
+task fetch_feeds_import: :environment do
+  Feeds::Import.call
+end
+
 # Temporary
 # @sre:mstruve This is temporary until we have an efficient way to handle this task
 # in Sidekiq for our large DEV community.

spec/fixtures/approvals/feeds_import/call/fetch_only_articles_from_a_feed_url.approved.txt

@@ -0,0 +1 @@
+21

spec/fixtures/approvals/feeds_import/call/parses_correctly.approved.txt

@@ -0,0 +1,31 @@
+---
+title: Testing RSS Feed
+published: false
+date: 2018-01-02 19:06:30 UTC
+tags: test
+canonical_url: 
+---
+
+youtube link here
+
+{% youtube QOCaacO8wus %}
+
+tweet here
+
+{% tweet 948256083352735744 %}
+
+Github gist here
+
+<iframe src="" width="0" height="0" frameborder="0" scrolling="no"><a href="https://medium.com/media/777ae8b7d8916e565b84b704c088cf0c/href">https://medium.com/media/777ae8b7d8916e565b84b704c088cf0c/href</a></iframe>
+
+code block
+
+```
+testsetsetsetsetsetset lets introduce some {{ chaos }}
+
+Here's more {{ what }}
+```
+
+some more code and `{{ VARIABLE }}` and **_`{{ HTML }}`_**
+
+### `{{ how about this }}`

spec/services/feeds/import_spec.rb

@@ -0,0 +1,130 @@
+require "rails_helper"
+
+RSpec.describe Feeds::Import, type: :service, vcr: true, db_strategy: :truncation do
+  self.use_transactional_tests = false
+
+  let(:link) { "https://medium.com/feed/@vaidehijoshi" }
+  let(:nonmedium_link) { "https://circleci.com/blog/feed.xml" }
+  let(:nonpermanent_link) { "https://medium.com/feed/@macsiri/" }
+
+  describe ".call" do
+    before do
+      [link, nonmedium_link, nonpermanent_link].each do |feed_url|
+        create(:user, feed_url: feed_url)
+      end
+    end
+
+    it "fetch only articles from a feed_url", vcr: { cassette_name: "feeds_import" } do
+      num_articles = described_class.call
+
+      verify(format: :txt) { num_articles }
+    end
+
+    it "does not recreate articles if they already exist", vcr: { cassette_name: "feeds_import_twice" } do
+      described_class.call
+
+      expect { described_class.call }.not_to change(Article, :count)
+    end
+
+    it "parses correctly", vcr: { cassette_name: "rss_reader_fetch_articles" } do
+      described_class.call
+
+      verify format: :txt do
+        User.find_by(feed_url: nonpermanent_link).articles.first.body_markdown
+      end
+    end
+
+    it "sets feed_fetched_at to the current time", vcr: { cassette_name: "feeds_import" } do
+      Timecop.freeze(Time.current) do
+        described_class.call
+
+        user = User.find_by(feed_url: nonpermanent_link)
+        feed_fetched_at = user.feed_fetched_at
+        expect(feed_fetched_at.to_i).to eq(Time.current.to_i)
+      end
+    end
+
+    it "does refetch same user over and over by default", vcr: { cassette_name: "feeds_import_multiple_times" } do
+      user = User.find_by(feed_url: nonpermanent_link)
+
+      Timecop.freeze(Time.current) do
+        user.update_columns(feed_fetched_at: Time.current)
+
+        fetched_at_time = user.reload.feed_fetched_at
+
+        # travel a few seconds in the future to simulate a new time
+        3.times do |i|
+          Timecop.travel((i + 5).seconds.from_now) do
+            described_class.call
+          end
+        end
+
+        expect(user.reload.feed_fetched_at > fetched_at_time).to be(true)
+      end
+    end
+
+    # it "reports an article creation error" do
+    #   allow(described_classs).to receive(:create_articles_from_user_feed).and_raise(StandardError)
+    #   allow(Honeybadger).to receive(:notify)
+
+    #   described_class.call
+
+    #   expect(Honeybadger).to have_received(:notify).at_least(:once)
+    # end
+
+    # it "reports a fetching error" do
+    #   allow(rss_reader).to receive(:fetch_feeds).and_raise(StandardError)
+    #   allow(Honeybadger).to receive(:notify)
+
+    #   described_class.call
+
+    #   expect(Honeybadger).to have_received(:notify).at_least(:once)
+    # end
+
+    it "queues as many slack messages as there are articles", vcr: { cassette_name: "feeds_import" } do
+      old_count = Slack::Messengers::Worker.jobs.count
+      num_articles = described_class.call
+      expect(Slack::Messengers::Worker.jobs.count).to eq(old_count + num_articles)
+    end
+  end
+
+  context "when feed_referential_link is false" do
+    it "does not self-reference links for user" do
+      # Article.find_by is used by find_and_replace_possible_links!
+      # checking its invocation is a shortcut to testing the functionality.
+      allow(Article).to receive(:find_by).and_call_original
+
+      create(:user, feed_url: nonpermanent_link, feed_referential_link: false)
+
+      described_class.call
+
+      expect(Article).not_to have_received(:find_by)
+    end
+  end
+
+  # describe "feeds parsing and regressions" do
+  #   it "parses https://medium.com/feed/@dvirsegal correctly", vcr: { cassette_name: "feeds_import_dvirsegal" } do
+  #     user = create(:user, feed_url: "https://medium.com/feed/@dvirsegal")
+
+  #     expect do
+  #       rss_reader.fetch_user(user)
+  #     end.to change(user.articles, :count).by(10)
+  #   end
+
+  #   it "converts/replaces <picture> tags to <img>", vcr: { cassette_name: "feeds_import_swimburger" } do
+  #     user = create(:user, feed_url: "https://swimburger.net/atom.xml")
+
+  #     expect do
+  #       rss_reader.fetch_user(user)
+  #     end.to change(user.articles, :count).by(10)
+
+  #     body_markdown = user.articles.last.body_markdown
+
+  #     expect(body_markdown).not_to include("<picture>")
+  #     expected_image_markdown =
+  #       "![Screenshot of Azure left navigation pane](https://swimburger.net/media/lxypkhak/azure-create-a-resource.png)"
+
+  #     expect(body_markdown).to include(expected_image_markdown)
+  #   end
+  # end
+end

spec/services/rss_reader_spec.rb

@@ -7,7 +7,6 @@ RSpec.describe RssReader, type: :service, vcr: true, db_strategy: :truncation do
   let(:link) { "https://medium.com/feed/@vaidehijoshi" }
   let(:nonmedium_link) { "https://circleci.com/blog/feed.xml" }
   let(:nonpermanent_link) { "https://medium.com/feed/@macsiri/" }
-  let(:rss_data) { RSS::Parser.parse(HTTParty.get(link).body, false) }
   let!(:rss_reader) { described_class.new }
 
   describe "#get_all_articles" do

spec/workers/articles/dev_feeds_import_worker_spec.rb

@@ -0,0 +1,39 @@
+require "rails_helper"
+
+RSpec.describe Articles::DevFeedsImportWorker, type: :worker do
+  let(:worker) { subject }
+
+  include_examples "#enqueues_on_correct_queue", "medium_priority"
+
+  describe "#perform" do
+    it "does not call the Feeds::Import for non DEV communities" do
+      allow(SiteConfig).to receive(:community_name).and_return("NotDEV")
+      allow(Feeds::Import).to receive(:call)
+
+      worker.perform
+
+      expect(Feeds::Import).not_to have_received(:call)
+    end
+
+    it "does not call the RssReader if the cache instructs it to cancel" do
+      allow(SiteConfig).to receive(:community_name).and_return("DEV")
+      allow(Rails.cache).to receive(:read).with("cancel_feeds_import").and_return("true")
+
+      allow(Feeds::Import).to receive(:call)
+
+      worker.perform
+
+      expect(Feeds::Import).not_to have_received(:call)
+    end
+
+    it "calls the RssReader to get all articles" do
+      allow(SiteConfig).to receive(:community_name).and_return("DEV")
+      allow(Rails.cache).to receive(:read).with("cancel_feeds_import").and_return(nil)
+      allow(Feeds::Import).to receive(:call)
+
+      worker.perform
+
+      expect(Feeds::Import).to have_received(:call)
+    end
+  end
+end