Feeds::Import: only import feeds that haven't been recently updated (#12126)

* Add earlier_than to Feeds::Import

* Ignore feeds that have been imported in the last 4 hours

* Make earlier_than mandatory in Feeds::ImportArticlesWorker

app/controllers/users_controller.rb

@@ -336,9 +336,11 @@ class UsersController < ApplicationController
   def import_articles_from_feed(user)
     return if user.feed_url.blank?
 
-    worker = FeatureFlag.enabled?(:feeds_import) ? Feeds::ImportArticlesWorker : RssReaderFetchUserWorker
-
-    worker.perform_async(user.id)
+    if FeatureFlag.enabled?(:feeds_import)
+      Feeds::ImportArticlesWorker.perform_async(nil, user.id)
+    else
+      RssReaderFetchUserWorker.perform_async(user.id)
+    end
   end
 
   def profile_params

app/services/feeds/import.rb

@@ -1,12 +1,13 @@
 module Feeds
   class Import
-    def self.call(users: nil)
-      new(users: users).call
+    def self.call(users: nil, earlier_than: nil)
+      new(users: users, earlier_than: earlier_than).call
     end
 
-    def initialize(users: nil)
+    def initialize(users: nil, earlier_than: nil)
       # using nil here to avoid an unnecessary table count to check presence
       @users = users || User.with_feed
+      @earlier_than = earlier_than
 
       # NOTE: should these be configurable? Currently they are the result of empiric
       # tests trying to find a balance between memory occupation and speed
@@ -40,6 +41,9 @@ module Feeds
         total_articles_count += articles.length
 
         articles.each { |article| Slack::Messengers::ArticleFetchedFeed.call(article: article) }
+
+        # we use `feed_fetched_at` to mark the last time a particular user's feed has been fetched, parsed and imported
+        batch_of_users.update_all(feed_fetched_at: Time.current)
       end
 
       DatadogStatsClient.count("feeds::import::articles.count", total_articles_count)
@@ -48,7 +52,15 @@ module Feeds
 
     private
 
-    attr_reader :users, :users_batch_size, :num_fetchers, :num_parsers
+    attr_reader :earlier_than, :users_batch_size, :num_fetchers, :num_parsers
+
+    def users
+      return @users unless earlier_than
+
+      # Filtering users whose feed hasn't been processed in the last `earlier_than` time span.
+      # New users + any user whose feed was processed earlier than the given time
+      @users.where(feed_fetched_at: nil).or(@users.where(feed_fetched_at: ..earlier_than))
+    end
 
     # TODO: put this in separate service object
     def fetch_feeds(batch_of_users)
@@ -79,8 +91,6 @@ module Feeds
         next
       end
 
-      batch_of_users.update_all(feed_fetched_at: Time.current)
-
       result.compact.to_h
     end
 

app/workers/articles/rss_reader_worker.rb

@@ -11,7 +11,7 @@ module Articles
       return if SiteConfig.dev_to?
 
       if FeatureFlag.enabled?(:feeds_import)
-        ::Feeds::ImportArticlesWorker.perform_async
+        ::Feeds::ImportArticlesWorker.perform_async(4.hours.ago)
       else
         # don't force fetch. Fetch "random" subset instead of all of them.
         ::RssReader.get_all_articles(force: false)

app/workers/feeds/import_articles_worker.rb

@@ -4,16 +4,10 @@ module Feeds
 
     sidekiq_options queue: :medium_priority, retry: 10
 
-    def perform(user_ids = [])
-      user_ids = Array.wrap(user_ids)
+    def perform(earlier_than, user_ids = [])
+      users = user_ids.present? ? User.where(id: user_ids) : nil
 
-      users = if user_ids.present?
-                User.where(id: user_ids)
-              else
-                User.with_feed
-              end
-
-      ::Feeds::Import.call(users: users)
+      ::Feeds::Import.call(users: users, earlier_than: earlier_than)
     end
   end
 end

lib/tasks/fetch.rake

@@ -2,7 +2,7 @@ desc "This task is called by the Heroku scheduler add-on"
 
 task fetch_all_rss: :environment do
   if FeatureFlag.enabled?(:feeds_import)
-    Feeds::Import.call
+    Feeds::Import.call(earlier_than: 4.hours.ago)
   else
     # don't force fetch. Fetch "random" subset instead of all of them.
     RssReader.get_all_articles(force: false)
@@ -10,7 +10,7 @@ task fetch_all_rss: :environment do
 end
 
 task fetch_feeds_import: :environment do
-  Feeds::Import.call
+  Feeds::Import.call(earlier_than: 4.hours.ago)
 end
 
 # Temporary

spec/requests/user/user_settings_spec.rb

@@ -309,7 +309,7 @@ RSpec.describe "UserSettings", type: :request do
       let(:user) { create(:user, feed_url: feed_url) }
 
       it "invokes RssReaderFetchUserWorker" do
-        allow(Feeds::ImportArticlesWorker).to receive(:perform_async).with(user.id)
+        allow(Feeds::ImportArticlesWorker).to receive(:perform_async).with(nil, user.id)
         allow(RssReaderFetchUserWorker).to receive(:perform_async).with(user.id)
 
         put user_path(user.id), params: { user: { feed_url: feed_url } }
@@ -319,13 +319,13 @@ RSpec.describe "UserSettings", type: :request do
       end
 
       it "invokes Feeds::ImportArticlesWorker if feeds_import feature flag is on" do
-        allow(Feeds::ImportArticlesWorker).to receive(:perform_async).with(user.id)
+        allow(Feeds::ImportArticlesWorker).to receive(:perform_async).with(nil, user.id)
         allow(RssReaderFetchUserWorker).to receive(:perform_async).with(user.id)
         allow(FeatureFlag).to receive(:enabled?).with(:feeds_import).and_return(true)
 
         put user_path(user.id), params: { user: { feed_url: feed_url } }
 
-        expect(Feeds::ImportArticlesWorker).to have_received(:perform_async).with(user.id)
+        expect(Feeds::ImportArticlesWorker).to have_received(:perform_async).with(nil, user.id)
         expect(RssReaderFetchUserWorker).not_to have_received(:perform_async).with(user.id)
       end
     end

spec/services/feeds/import_spec.rb

@@ -44,25 +44,6 @@ RSpec.describe Feeds::Import, type: :service, vcr: true, db_strategy: :truncatio
       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 "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
@@ -123,8 +104,61 @@ RSpec.describe Feeds::Import, type: :service, vcr: true, db_strategy: :truncatio
     end
   end
 
+  context "when refetching" do
+    before do
+      [link, nonmedium_link, nonpermanent_link].each do |feed_url|
+        create(:user, feed_url: feed_url)
+      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 "does not refetch recently fetched users if earlier_than is given", vcr: { cassette_name: "feeds_import" } do
+      time = 30.minutes.ago
+
+      Timecop.freeze(time) do
+        described_class.call
+      end
+
+      # we delete the articles to make sure it won't trigger the duplicate check
+      Article.delete_all
+
+      expect { described_class.call(earlier_than: 1.hour.ago) }.not_to change(Article, :count)
+    end
+
+    it "refetches recently fetched users if earlier_than is now", vcr: { cassette_name: "feeds_import_twice" } do
+      time = 30.minutes.ago
+
+      Timecop.freeze(time) do
+        described_class.call
+      end
+
+      # we delete the articles to make sure it won't trigger the duplicate check
+      Article.delete_all
+
+      expect { described_class.call(earlier_than: Time.current) }.to change(Article, :count)
+    end
+  end
+
   context "when feed_referential_link is false" do
-    it "does not self-reference links for user" do
+    it "does not self-reference links for user", vcr: { cassette_name: "feeds_import_non_referential" } 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

spec/workers/articles/rss_reader_worker_spec.rb

@@ -25,14 +25,16 @@ RSpec.describe Articles::RssReaderWorker, type: :worker do
     end
 
     it "enqueues Feeds::ImportArticlesWorker if the :feeds_import flag is enabled" do
+      allow(Feeds::ImportArticlesWorker).to receive(:perform_async)
       allow(RssReader).to receive(:get_all_articles)
       allow(FeatureFlag).to receive(:enabled?).with(:feeds_import).and_return(true)
 
-      sidekiq_assert_enqueued_jobs(1, only: Feeds::ImportArticlesWorker) do
+      Timecop.freeze(Time.current) do
         worker.perform
-      end
 
-      expect(RssReader).not_to have_received(:get_all_articles)
+        expect(RssReader).not_to have_received(:get_all_articles)
+        expect(Feeds::ImportArticlesWorker).to have_received(:perform_async).with(4.hours.ago)
+      end
     end
 
     it "short circuits if it's running on DEV" do

spec/workers/feeds/import_articles_worker_spec.rb

@@ -9,7 +9,7 @@ RSpec.describe Feeds::ImportArticlesWorker, type: :worker do
     it "calls the Feeds::Import to get all articles" do
       allow(Feeds::Import).to receive(:call)
 
-      worker.perform
+      worker.perform(1.hour.ago)
 
       expect(Feeds::Import).to have_received(:call)
     end
@@ -19,9 +19,21 @@ RSpec.describe Feeds::ImportArticlesWorker, type: :worker do
         user = create(:user)
         allow(Feeds::Import).to receive(:call)
 
-        worker.perform([user.id])
+        worker.perform(nil, [user.id])
 
-        expect(Feeds::Import).to have_received(:call).with(users: User.where(id: [user.id]))
+        expect(Feeds::Import).to have_received(:call).with(users: User.where(id: [user.id]), earlier_than: nil)
+      end
+    end
+
+    context "with earlier_than time" do
+      it "calls Feeds::Import with the correct time if given" do
+        allow(Feeds::Import).to receive(:call)
+
+        Timecop.freeze(Time.current) do
+          worker.perform(4.hours.ago)
+
+          expect(Feeds::Import).to have_received(:call).with(users: nil, earlier_than: 4.hours.ago)
+        end
       end
     end
   end