fix sorting events in rom integration

Author: pjurewicz

contrib/ruby_event_store-rom/lib/ruby_event_store/rom/relations/events.rb

@@ -34,20 +34,36 @@ def by_event_type(types)
           where(event_type: types)
         end
 
-        def newer_than(time)
-          where { |r| r.events[:created_at] > time.localtime }
+        def newer_than(time, time_sort_by)
+          if time_sort_by == :as_of
+            where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) > time.localtime }
+          else
+            where { |r| r.events[:created_at] > time.localtime }
+          end
         end
 
-        def newer_than_or_equal(time)
-          where { |r| r.events[:created_at] >= time.localtime }
+        def newer_than_or_equal(time, time_sort_by)
+          if time_sort_by == :as_of
+            where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) >= time.localtime }
+          else
+            where { |r| r.events[:created_at] >= time.localtime }
+          end
         end
 
-        def older_than(time)
-          where { |r| r.events[:created_at] < time.localtime }
+        def older_than(time, time_sort_by)
+          if time_sort_by == :as_of
+            where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) < time.localtime }
+          else
+            where { |r| r.events[:created_at] < time.localtime }
+          end
         end
 
-        def older_than_or_equal(time)
-          where { |r| r.events[:created_at] <= time.localtime }
+        def older_than_or_equal(time, time_sort_by)
+          if time_sort_by == :as_of
+            where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) <= time.localtime }
+          else
+            where { |r| r.events[:created_at] <= time.localtime }
+          end
         end
 
         DIRECTION_MAP = { forward: %i[asc > <], backward: %i[desc < >] }.freeze

contrib/ruby_event_store-rom/lib/ruby_event_store/rom/relations/stream_entries.rb

@@ -34,20 +34,36 @@ def max_position(stream)
           by_stream(stream).select(:position).order(Sequel.desc(:position)).first
         end
 
-        def newer_than(time)
-          join_events.where { |r| r.events[:created_at] > time.localtime }
+        def newer_than(time, time_sort_by)
+          if time_sort_by == :as_of
+            join_events.where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) > time.localtime }
+          else
+            join_events.where { |r| r.events[:created_at] > time.localtime }
+          end
         end
 
-        def newer_than_or_equal(time)
-          join_events.where { |r| r.events[:created_at] >= time.localtime }
+        def newer_than_or_equal(time, time_sort_by)
+          if time_sort_by == :as_of
+            join_events.where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) >= time.localtime }
+          else
+            join_events.where { |r| r.events[:created_at] >= time.localtime }
+          end
         end
 
-        def older_than(time)
-          join_events.where { |r| r.events[:created_at] < time.localtime }
+        def older_than(time, time_sort_by)
+          if time_sort_by == :as_of
+            join_events.where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) < time.localtime }
+          else
+            join_events.where { |r| r.events[:created_at] < time.localtime }
+          end
         end
 
-        def older_than_or_equal(time)
-          join_events.where { |r| r.events[:created_at] <= time.localtime }
+        def older_than_or_equal(time, time_sort_by)
+          if time_sort_by == :as_of
+            join_events.where { |r| string::coalesce(r.events[:valid_at], r.events[:created_at]) <= time.localtime }
+          else
+            join_events.where { |r| r.events[:created_at] <= time.localtime }
+          end
         end
 
         DIRECTION_MAP = { forward: %i[asc > <], backward: %i[desc < >] }.freeze

contrib/ruby_event_store-rom/lib/ruby_event_store/rom/repositories/events.rb

@@ -110,10 +110,10 @@ def read_scope(specification)
 
           query = query.by_event_id(specification.with_ids) if specification.with_ids
           query = query.by_event_type(specification.with_types) if specification.with_types?
-          query = query.older_than(specification.older_than) if specification.older_than
-          query = query.older_than_or_equal(specification.older_than_or_equal) if specification.older_than_or_equal
-          query = query.newer_than(specification.newer_than) if specification.newer_than
-          query = query.newer_than_or_equal(specification.newer_than_or_equal) if specification.newer_than_or_equal
+          query = query.older_than(specification.older_than, specification.time_sort_by) if specification.older_than
+          query = query.older_than_or_equal(specification.older_than_or_equal, specification.time_sort_by) if specification.older_than_or_equal
+          query = query.newer_than(specification.newer_than, specification.time_sort_by) if specification.newer_than
+          query = query.newer_than_or_equal(specification.newer_than_or_equal, specification.time_sort_by) if specification.newer_than_or_equal
           query
         end