Move record id preparation into the JSON ingestion adapter

- The indexer copied the envelope `id` into the record for every format.
  Only the JSON adapter knows that `record` is a Hash, so the copy belongs
  there. This removes the last format-specific `is_a?` check from the
  operation factory.
- `ValidationResult` now carries the event to build operations from, so an
  adapter can return a prepared copy of the event.
- Prepares for a typed `Event` whose `record` type is a per-format
  generic parameter.
- Narrows `validation_result.record_preparer`/`.event` with `# : ...` type
  comments instead of `or raise`: both are always present together by
  construction once `failure` is nil, so the assertion can't fire.

Generated with Claude Code

Author: myronmarston

elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb

@@ -17,13 +17,15 @@ module IngestionAdapter
       class Interface
         # Validates the given event and resolves the record preparer appropriate for the event's
         # ingestion schema version. The indexer selects this adapter from the event's format tag.
+        # The indexer builds operations from the event in the returned result, so an adapter can
+        # return a prepared copy of the event.
         #
         # @param event [Hash<String, Object>] an ElasticGraph indexing event
         # @param skip_record_validation [Boolean] whether to skip record validation; the event envelope must still be validated
         # @return [ValidationResult] the result of validating the event
         def validate_event(event, skip_record_validation: false)
           # simplecov:disable -- must return a result to satisfy Steep type checking but never called
-          ValidationResult.valid(RecordPreparer::Identity)
+          ValidationResult.valid(event, RecordPreparer::Identity)
           # simplecov:enable
         end
       end
@@ -39,19 +41,22 @@ def validate_event(event, skip_record_validation: false)
       # Returned by {Interface#validate_event}. A non-nil `failure` indicates an invalid event,
       # and a non-nil `record_preparer` indicates a valid event.
       #
+      # @!attribute [r] event
+      #   @return [Hash<String, Object>, nil] the event to build operations from, when the event is valid
       # @!attribute [r] record_preparer
       #   @return [Object, nil] preparer for the event's record, when the event is valid
       # @!attribute [r] failure
       #   @return [Failure, nil] description of the validation problem, when the event is invalid
-      ValidationResult = ::Data.define(:record_preparer, :failure) do
+      ValidationResult = ::Data.define(:event, :record_preparer, :failure) do
         # @implements ValidationResult
 
         # Builds a result for a valid event.
         #
+        # @param event [Hash<String, Object>] the event to build operations from; an adapter may return a prepared copy
         # @param record_preparer [Object] preparer for the event's record
         # @return [ValidationResult]
-        def self.valid(record_preparer)
-          new(record_preparer: record_preparer, failure: nil)
+        def self.valid(event, record_preparer)
+          new(event: event, record_preparer: record_preparer, failure: nil)
         end
 
         # Builds a result for an invalid event.
@@ -60,7 +65,7 @@ def self.valid(record_preparer)
         # @param message [String] detailed validation failure message
         # @return [ValidationResult]
         def self.invalid(validation_target:, message:)
-          new(record_preparer: nil, failure: Failure.new(validation_target: validation_target, message: message))
+          new(event: nil, record_preparer: nil, failure: Failure.new(validation_target: validation_target, message: message))
         end
       end
     end

elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb

@@ -25,7 +25,6 @@ class Factory < Support::MemoizableData.define(
         :skip_record_validation_percents_by_type
       )
         def build(event)
-          event = prepare_event(event)
           format = event.fetch(INGESTION_FORMAT_KEY, "json")
           adapter = ingestion_adapters_by_format[format]
 
@@ -50,7 +49,8 @@ def build_for_adapter(event, adapter)
           if (failure = validation_result.failure)
             build_failed_result(event, failure.validation_target, failure.message)
           else
-            record_preparer = validation_result.record_preparer or raise
+            record_preparer = validation_result.record_preparer # : _RecordPreparer
+            event = validation_result.event # : event
             if skip_record_validation
               build_success_result_isolating_malformed_records(event, record_preparer, adapter)
             else
@@ -85,13 +85,6 @@ def build_success_result_isolating_malformed_records(event, record_preparer, ada
           build_failed_result(event, failure.validation_target, failure.message)
         end
 
-        # This copies the `id` from event into the actual record
-        # This is necessary because we want to index `id` as part of the record so that the datastore will include `id` in returned search payloads.
-        def prepare_event(event)
-          return event unless event["record"].is_a?(::Hash) && event["id"]
-          event.merge("record" => event["record"].merge("id" => event.fetch("id")))
-        end
-
         # `Zlib.crc32` returns a value in `[0, 2**32)`. Pre-dividing that space by 100 lets us test a
         # configured percent with a single multiply instead of dividing on every event.
         CRC32_SPACE_PER_PERCENT = (1 << 32) / 100.0

elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs

@@ -22,16 +22,17 @@ module ElasticGraph
       end
 
       class ValidationResultSupertype < ::Data
+        attr_reader event: event?
         attr_reader record_preparer: _RecordPreparer?
         attr_reader failure: Failure?
 
-        def initialize: (record_preparer: _RecordPreparer?, failure: Failure?) -> void
-        def self.new: (record_preparer: _RecordPreparer?, failure: Failure?) -> instance
+        def initialize: (event: event?, record_preparer: _RecordPreparer?, failure: Failure?) -> void
+        def self.new: (event: event?, record_preparer: _RecordPreparer?, failure: Failure?) -> instance
         def self.members: () -> ::Array[::Symbol]
       end
 
       class ValidationResult < ValidationResultSupertype
-        def self.valid: (_RecordPreparer) -> ValidationResult
+        def self.valid: (event, _RecordPreparer) -> ValidationResult
         def self.invalid: (validation_target: ::String, message: ::String) -> ValidationResult
       end
     end

elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs

@@ -36,7 +36,6 @@ module ElasticGraph
         private
 
         def build_for_adapter: (event, IngestionAdapter::_Adapter) -> BuildResult
-        def prepare_event: (event) -> event
         def build_success_result: (event, _RecordPreparer, type_with_skipped_validation: ::String?) -> BuildResult
         def build_success_result_isolating_malformed_records: (event, _RecordPreparer, IngestionAdapter::_Adapter) -> BuildResult
         CRC32_SPACE_PER_PERCENT: ::Float

elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb

@@ -470,16 +470,19 @@ def factory_whose_record_preparation_is_broken
             end
           end
 
-          it "uses a registered non-JSON adapter for events in that alternate format" do
+          it "uses a registered non-JSON adapter for events in that alternate format, building operations from the event the adapter returns" do
             event = build_upsert_event(:component, id: "1", __version: 1).merge(INGESTION_FORMAT_KEY => "other")
+            prepared_event = event.merge("record" => event.fetch("record").merge("name" => "prepared by adapter"))
             other_adapter = instance_spy(
               IngestionAdapter::Interface,
-              validate_event: IngestionAdapter::ValidationResult.valid(RecordPreparer::Identity)
+              validate_event: IngestionAdapter::ValidationResult.valid(prepared_event, RecordPreparer::Identity)
             )
             factory = indexer.operation_factory.with(ingestion_adapters_by_format: {"other" => other_adapter})
 
-            expect(factory.build(event).operations).not_to be_empty
+            operations = factory.build(event).operations
 
+            expect(operations.size).to eq 1
+            expect(operations.map(&:event)).to eq [prepared_event]
             expect(other_adapter).to have_received(:validate_event).with(event, skip_record_validation: false)
           end
 

elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb

@@ -52,14 +52,15 @@ def validate_event(event, skip_record_validation: false)
           return ValidationResult.invalid(validation_target: "event payload", message: error_message)
         end
 
-        record = event.fetch("record")
+        # The datastore includes `id` in search payloads only when it is part of the indexed record.
+        record = event.fetch("record").merge("id" => event.fetch("id"))
         graphql_type_name = event.fetch("type")
 
         if !skip_record_validation && (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record))
           return ValidationResult.invalid(validation_target: "#{graphql_type_name} record", message: error_message)
         end
 
-        ValidationResult.valid(@record_preparer_factory.for_json_schema_version(selected_json_schema_version))
+        ValidationResult.valid(event.merge("record" => record), @record_preparer_factory.for_json_schema_version(selected_json_schema_version))
       end
 
       private

elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb

@@ -25,6 +25,18 @@ module JSONIngestion
           expect(result.operations.map(&:doc_id)).to include("1")
         end
 
+        it "copies the envelope `id` into the record so that the datastore includes `id` in search payloads" do
+          event = build_upsert_event(:component, id: "1")
+          event.fetch("record").delete("id")
+
+          result = indexer.operation_factory.build(event)
+
+          expect(result.failed_event_error).to be nil
+          expect(result.operations.size).to eq 1
+          expect(result.operations.map(&:prepared_record)).to all include("id" => "1")
+          expect(event.fetch("record")).to exclude("id")
+        end
+
         # The factory replaces an adapter's missing preparer with an identity preparer while
         # building its failed result, so this adapter contract must be asserted directly.
         it "returns no record preparer, plus the part of the event at fault, for an event it rejects" do