Merge pull request #165 from cantino/prevent_event_replay_on_linking

Prevent event propagation of events created before a link

app/models/agent.rb

@@ -252,7 +252,7 @@ def receive!
                 select("agents.id AS receiver_agent_id, sources.id AS source_agent_id, events.id AS event_id").
                 joins("JOIN links ON (links.receiver_id = agents.id)").
                 joins("JOIN agents AS sources ON (links.source_id = sources.id)").
-                joins("JOIN events ON (events.agent_id = sources.id)").
+                joins("JOIN events ON (events.agent_id = sources.id AND events.id > links.event_id_at_creation)").
                 where("agents.last_checked_event_id IS NULL OR events.id > agents.last_checked_event_id").to_sql
 
         agents_to_events = {}

app/models/link.rb

@@ -4,4 +4,10 @@ class Link < ActiveRecord::Base
 
   belongs_to :source, :class_name => "Agent", :inverse_of => :links_as_source
   belongs_to :receiver, :class_name => "Agent", :inverse_of => :links_as_receiver
+
+  before_create :store_event_id_at_creation
+
+  def store_event_id_at_creation
+    self.event_id_at_creation = source.events.limit(1).reorder("id desc").pluck(:id).first || 0
+  end
 end

db/migrate/20140213053001_add_event_id_at_creation_to_links.rb

@@ -0,0 +1,18 @@
+class AddEventIdAtCreationToLinks < ActiveRecord::Migration
+  def up
+    add_column :links, :event_id_at_creation, :integer, :null => false, :default => 0
+
+    execute <<-SQL
+      UPDATE #{ActiveRecord::Base.connection.quote_table_name('links')}
+      SET event_id_at_creation = (
+        SELECT #{ActiveRecord::Base.connection.quote_column_name('id')}
+        FROM #{ActiveRecord::Base.connection.quote_table_name('events')}
+        WHERE events.agent_id = links.source_id ORDER BY events.id DESC limit 1
+      )
+    SQL
+  end
+
+  def down
+    remove_column :links, :event_id_at_creation
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended to check this file into your version control system.
 
-ActiveRecord::Schema.define(:version => 20140210062747) do
+ActiveRecord::Schema.define(:version => 20140213053001) do
 
   create_table "agent_logs", :force => true do |t|
     t.integer  "agent_id",                                             :null => false
@@ -88,8 +88,9 @@
   create_table "links", :force => true do |t|
     t.integer  "source_id"
     t.integer  "receiver_id"
-    t.datetime "created_at",  :null => false
-    t.datetime "updated_at",  :null => false
+    t.datetime "created_at",                          :null => false
+    t.datetime "updated_at",                          :null => false
+    t.integer  "event_id_at_creation", :default => 0, :null => false
   end
 
   add_index "links", ["receiver_id", "source_id"], :name => "index_links_on_receiver_id_and_source_id"

spec/models/agent_spec.rb

@@ -221,8 +221,13 @@ def receive(events)
       it "should track when events have been seen and not received them again" do
         mock.any_instance_of(Agents::TriggerAgent).receive(anything).once
         Agent.async_check(agents(:bob_weather_agent).id)
-        Agent.receive!
-        Agent.receive!
+        lambda {
+          Agent.receive!
+        }.should change { agents(:bob_rain_notifier_agent).reload.last_checked_event_id }
+
+        lambda {
+          Agent.receive!
+        }.should_not change { agents(:bob_rain_notifier_agent).reload.last_checked_event_id }
       end
 
       it "should not run consumers that have nothing to do" do
@@ -238,6 +243,39 @@ def receive(events)
         Agent.async_check(agents(:jane_weather_agent).id)
         Agent.receive!
       end
+
+      it "should ignore events that were created before a particular Link" do
+        agent2 = Agents::SomethingSource.new(:name => "something")
+        agent2.user = users(:bob)
+        agent2.save!
+        agent2.check
+
+        mock.any_instance_of(Agents::TriggerAgent).receive(anything).twice
+        agents(:bob_weather_agent).check # bob_weather_agent makes an event
+
+        lambda {
+          Agent.receive! # event gets propagated
+        }.should change { agents(:bob_rain_notifier_agent).reload.last_checked_event_id }
+
+        # This agent creates a few events before we link to it, but after our last check.
+        agent2.check
+        agent2.check
+
+        # Now we link to it.
+        agents(:bob_rain_notifier_agent).sources << agent2
+        agent2.links_as_source.first.event_id_at_creation.should == agent2.events.reorder("events.id desc").first.id
+
+        lambda {
+          Agent.receive! # but we don't receive those events because they're too old
+        }.should_not change { agents(:bob_rain_notifier_agent).reload.last_checked_event_id }
+
+        # Now a new event is created by agent2
+        agent2.check
+
+        lambda {
+          Agent.receive! # and we receive it
+        }.should change { agents(:bob_rain_notifier_agent).reload.last_checked_event_id }
+      end
     end
 
     describe "creating a new agent and then calling .receive!" do