Optimize valid_at

Author: lukaszreszke

contrib/ruby_event_store-sequel/lib/ruby_event_store/sequel/event_repository.rb

@@ -27,7 +27,7 @@ def append_to_stream(records, stream, expected_version)
               data: serialized_record.data,
               metadata: serialized_record.metadata,
               created_at: record.timestamp,
-              valid_at: record.valid_at
+              valid_at: optimize_timestamp(record.valid_at, record.timestamp)
             )
             unless stream.global?
               @db[:event_store_events_in_streams].insert(
@@ -102,7 +102,7 @@ def last_stream_event(stream)
             data: event[:data],
             metadata: event[:metadata],
             timestamp: event[:created_at].iso8601(TIMESTAMP_PRECISION),
-            valid_at: event[:valid_at].iso8601(TIMESTAMP_PRECISION)
+            valid_at: (event[:valid_at] || event[:created_at]).iso8601(TIMESTAMP_PRECISION)
           )
           .deserialize(@serializer)
       end
@@ -160,6 +160,10 @@ def streams_of(event_id)
 
       private
 
+      def optimize_timestamp(valid_at, created_at)
+        valid_at unless valid_at.eql?(created_at)
+      end
+
       def record(h)
         SerializedRecord
           .new(
@@ -304,7 +308,7 @@ def read_from_global_stream(specification)
         end
 
         dataset = dataset.order(:created_at) if specification.time_sort_by_as_at?
-        dataset = dataset.order(:valid_at) if specification.time_sort_by_as_of?
+        dataset = dataset.order(::Sequel.lit(coalesced_date)) if specification.time_sort_by_as_of?
         dataset = dataset.limit(specification.limit) if specification.limit?
         dataset = dataset.order(::Sequel[:event_store_events][:id]) unless specification.time_sort_by
         dataset = dataset.reverse if specification.backward?

contrib/ruby_event_store-sequel/spec/event_repository_spec.rb

@@ -96,6 +96,21 @@ module Sequel
         expect(record.valid_at).to eq(t)
       end
 
+      specify "valid-at storage optimization when same as created-at" do
+        repository.append_to_stream(
+          [SRecord.new(timestamp: time = with_precision(Time.at(0)))],
+          Stream.new(GLOBAL_STREAM),
+          ExpectedVersion.any
+        )
+        record = repository.read(specification.result).first
+        expect(record.timestamp).to eq(time)
+        expect(record.valid_at).to eq(time)
+
+        event_record = helper.sequel[:event_store_events].where(event_id: record.event_id).first
+        expect(event_record[:created_at]).to eq(time)
+        expect(event_record[:valid_at]).to be_nil
+      end
+
       private
 
       def with_precision(time)