Another case for backwards compatibility — bad configuration

Warn about data to migrate, while still allowing to read it with some
performance impact.

Author: mostlyobvious

ruby_event_store-active_record/lib/ruby_event_store/active_record/event_repository.rb

@@ -32,6 +32,33 @@ def initialize(model_factory: WithDefaultModels.new, serializer:)
           @event_klass.include(SkipJsonSerialization)
         end
         @repo_reader = EventRepositoryReader.new(@event_klass, @stream_klass, serializer)
+        EventRepositoryReader.prepend(
+          Module.new do
+            def unwrap(column_name, deserialized_payload)
+              case deserialized_payload
+              when String
+                warn "Double serialization of #{column_name} column detected"
+                JSON.parse(deserialized_payload)
+              else
+                deserialized_payload
+              end
+            end
+
+            def record(...)
+              record_ = super
+
+              Record.new(
+                event_id: record_.event_id,
+                metadata: unwrap("metadata", record_.metadata),
+                data: unwrap("data", record_.data),
+                event_type: record_.event_type,
+                timestamp: record_.timestamp,
+                valid_at: record_.valid_at,
+              )
+            end
+          end,
+        ) if serializer == JSON && json_data_type?
+
         @index_violation_detector = IndexViolationDetector.new(@event_klass.table_name, @stream_klass.table_name)
       end
 

ruby_event_store-active_record/spec/event_repository_spec.rb

@@ -311,6 +311,29 @@ module ActiveRecord
         }.not_to raise_error
       end
 
+      specify "JSON/B backwards compatibility — with mistakenly configured double serialization" do
+        skip unless %w[json jsonb].include?(ENV["DATA_TYPE"])
+
+        repository = EventRepository.new(serializer: JSON)
+        repository.append_to_stream(
+          [SRecord.new(data: JSON.dump({ "simulate" => "double" }))],
+          Stream.new("stream"),
+          ExpectedVersion.any,
+        )
+        repository.append_to_stream(
+          [SRecord.new(data: { "simulate" => "single" })],
+          Stream.new("stream"),
+          ExpectedVersion.any,
+        )
+
+        expect(repository.read(specification.backward.limit(1).result).first.data).to eq({ "simulate" => "single" })
+        expect {
+          expect(repository.read(specification.forward.limit(1).result).first.data).to eq({ "simulate" => "double" })
+        }.to output(<<~MSG).to_stderr
+          Double serialization of data column detected
+        MSG
+      end
+
       private
 
       def with_precision(time)