Let the snapshot subquery reference the outer table (#94)

* Let the snapshot subquery reference the outer table

* Add spec for loading multiple aggregates of which one us snapshotted

* Upgrade ruby

.ruby-version

@@ -1 +1 @@
-2.2.2
+2.3.4

.travis.yml

@@ -8,8 +8,8 @@ env:
     - GIT_COMMITTED_AT=$(if [ "$TRAVIS_PULL_REQUEST" == "false" ]; then git log -1 --pretty=format:%ct; else git log -1 --skip 1 --pretty=format:%ct; fi)
     - COVERAGE=true
 rvm:
-  - 2.1.5
   - 2.2.2
+  - 2.3.4
 matrix:
   exclude:
     - rvm: 2.1.5

lib/sequent/core/event_store.rb

@@ -72,13 +72,13 @@ def load_events_for_aggregates(aggregate_ids)
 
         events = event_record_class.connection.select_all(%Q{
 SELECT event_type, event_json
-  FROM #{quote_table_name event_record_class.table_name}
+  FROM #{quote_table_name event_record_class.table_name} AS o
 WHERE aggregate_id in (#{aggregate_ids.map{ |aggregate_id| quote(aggregate_id)}.join(",")})
-  AND sequence_number >= COALESCE((SELECT MAX(sequence_number)
-                                      FROM #{quote_table_name event_record_class.table_name}
-                                     WHERE event_type = #{quote snapshot_event_class.name}
-                                       AND aggregate_id in (#{aggregate_ids.map{ |aggregate_id| quote(aggregate_id)}.join(",")})), 0)
-  ORDER BY sequence_number ASC, (CASE event_type WHEN #{quote snapshot_event_class.name} THEN 0 ELSE 1 END) ASC
+AND sequence_number >= COALESCE((SELECT MAX(sequence_number)
+                                 FROM #{quote_table_name event_record_class.table_name} AS i
+                                 WHERE event_type = #{quote snapshot_event_class.name}
+                                   AND i.aggregate_id = o.aggregate_id), 0)
+ORDER BY sequence_number ASC, (CASE event_type WHEN #{quote snapshot_event_class.name} THEN 0 ELSE 1 END) ASC
 }).map! do |event_hash|
           deserialize_event(event_hash)
         end

spec/lib/sequent/core/aggregate_snapshotter_spec.rb

@@ -18,6 +18,8 @@ class MyAggregate < Sequent::Core::AggregateRoot; end
       Sequent::configuration.command_handlers = commands_handlers
     end
   end
+  let(:snapshot_threshold) { 1 }
+  let(:events) { [MyEvent.new(aggregate_id: aggregate_id, sequence_number: 1)] }
 
   before :each do
     Sequent::configuration.command_handlers << described_class.new
@@ -26,7 +28,7 @@ class MyAggregate < Sequent::Core::AggregateRoot; end
       [
         [
           Sequent::Core::EventStream.new(aggregate_type: 'MyAggregate', aggregate_id: aggregate_id, snapshot_threshold: 1),
-          [MyEvent.new(aggregate_id: aggregate_id, sequence_number: 1)]
+          events
         ]
       ]
     )
@@ -37,4 +39,29 @@ class MyAggregate < Sequent::Core::AggregateRoot; end
 
     expect(Sequent::Core::EventRecord.last.event_type).to eq Sequent::Core::SnapshotEvent.name
   end
+
+  context 'loads aggregates with snapshots' do
+    let(:snapshot_threshold) { 2 }
+    let(:events) { [MyEvent.new(aggregate_id: aggregate_id, sequence_number: 1), MyEvent.new(aggregate_id: aggregate_id, sequence_number: 2), MyEvent.new(aggregate_id: aggregate_id, sequence_number: 3)] }
+
+    let(:aggregate_id_2) { Sequent.new_uuid }
+
+    before :each do
+      event_store.commit_events(
+        Sequent::Core::CommandRecord.new,
+        [
+          [
+            Sequent::Core::EventStream.new(aggregate_type: 'MyAggregate', aggregate_id: aggregate_id_2, snapshot_threshold: 10),
+            [MyEvent.new(aggregate_id: aggregate_id_2, sequence_number: 1)]
+          ]
+        ]
+      )
+
+      Sequent.command_service.execute_commands(*take_snapshot)
+    end
+
+    it 'loads both events' do
+      expect(event_store.load_events_for_aggregates([aggregate_id, aggregate_id_2])).to have(2).items
+    end
+  end
 end