From 5833483d25f474b3789d20a61102f7467b81a2a7 Mon Sep 17 00:00:00 2001
From: Josh Wilson <joshuaw@squareup.com>
Date: Tue, 7 Jul 2026 15:37:04 -0500
Subject: [PATCH 1/3] Extract an ingestion adapter seam inside
 elasticgraph-indexer

Operation::Factory no longer performs JSON-schema validation itself;
instead it routes each event to an ingestion adapter, which validates
the event and supplies the version-appropriate record preparer. The
existing JSON validation logic moves behind the seam as
IngestionAdapter::JSONEvents, so behavior is unchanged (the existing
validation specs pass as-is). Multiple adapters dispatch by
handles_event?, with a single available adapter receiving all events.
---
 .../lib/elastic_graph/indexer.rb              |  15 +-
 .../lib/elastic_graph/indexer/event_id.rb     |   2 +-
 .../indexer/ingestion_adapter.rb              |  87 +++++++++++
 .../indexer/ingestion_adapter/json_events.rb  | 138 ++++++++++++++++++
 .../indexer/operation/factory.rb              | 128 ++++------------
 .../indexer/spec_support/event_matcher.rb     |   9 +-
 .../sig/elastic_graph/indexer.rbs             |   3 +
 .../indexer/ingestion_adapter.rbs             |  42 ++++++
 .../indexer/ingestion_adapter/json_events.rbs |  31 ++++
 .../indexer/operation/factory.rbs             |  27 +---
 .../indexer/operation/factory_spec.rb         |  83 ++++++++++-
 11 files changed, 435 insertions(+), 130 deletions(-)
 create mode 100644 elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
 create mode 100644 elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb
 create mode 100644 elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs
 create mode 100644 elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs

diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb
index 3bb1a34e6..d4f5d6fbf 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb
@@ -63,6 +63,16 @@ def record_preparer_factory
       end
     end
 
+    # The ingestion adapters available for processing events. For now, only the JSON events
+    # adapter is available; indexer extension modules will be able to contribute additional
+    # adapters as alternate ingestion formats are supported.
+    def ingestion_adapters
+      @ingestion_adapters ||= begin
+        require "elastic_graph/indexer/ingestion_adapter/json_events"
+        [IngestionAdapter::JSONEvents.new(schema_artifacts: schema_artifacts, logger: logger)]
+      end
+    end
+
     def processor
       @processor ||= begin
         require "elastic_graph/indexer/processor"
@@ -82,11 +92,10 @@ def operation_factory
         Operation::Factory.new(
           schema_artifacts: schema_artifacts,
           index_definitions_by_graphql_type: datastore_core.index_definitions_by_graphql_type,
-          record_preparer_factory: record_preparer_factory,
+          ingestion_adapters: ingestion_adapters,
           logger: datastore_core.logger,
           skip_derived_indexing_type_updates: config.skip_derived_indexing_type_updates,
-          skip_record_validation_percents_by_type: config.skip_record_validation_percents_by_type,
-          configure_record_validator: nil
+          skip_record_validation_percents_by_type: config.skip_record_validation_percents_by_type
         )
       end
     end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb
index 30d22caaf..755894f67 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb
@@ -26,7 +26,7 @@ def to_s
 
     # Steep weirdly expects them here...
     # @dynamic initialize, config, datastore_core, schema_artifacts, datastore_router, monotonic_clock
-    # @dynamic record_preparer_factory, processor, operation_factory, logger
+    # @dynamic record_preparer_factory, processor, operation_factory, ingestion_adapters, logger
     # @dynamic self.from_parsed_yaml
   end
 end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
new file mode 100644
index 000000000..e9a4d1c1d
--- /dev/null
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
@@ -0,0 +1,87 @@
+# Copyright 2024 - 2026 Block, Inc.
+#
+# Use of this source code is governed by an MIT-style
+# license that can be found in the LICENSE file or at
+# https://opensource.org/licenses/MIT.
+#
+# frozen_string_literal: true
+
+module ElasticGraph
+  class Indexer
+    # Namespace for ingestion adapters. An ingestion adapter teaches the indexer how to handle
+    # events of a particular ingestion format: it validates each event and provides the
+    # version-appropriate machinery to prepare the event's record for indexing.
+    module IngestionAdapter
+      # Defines the ingestion adapter interface. Adapter classes are not required to subclass this,
+      # but must implement these methods.
+      class Interface
+        # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts
+        # @param logger [Logger] the ElasticGraph logger
+        def initialize(schema_artifacts:, logger:)
+          # must be defined, but nothing to do
+        end
+
+        # Indicates whether this adapter recognizes the given event as one of its own. When multiple
+        # adapters are available, the indexer routes each event to the first adapter that returns
+        # `true`. (When exactly one adapter is available, it receives all events.)
+        #
+        # @param event [Hash<String, Object>] an ElasticGraph indexing event
+        # @return [Boolean] whether this adapter handles the event
+        def handles_event?(event)
+          # :nocov: -- must return a boolean to satisfy Steep type checking but never called
+          false
+          # :nocov:
+        end
+
+        # Validates the given event and resolves the record preparer appropriate for the event's
+        # schema version.
+        #
+        # @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)
+          # :nocov: -- must return a result to satisfy Steep type checking but never called
+          ValidationResult.valid(RecordPreparer::Identity)
+          # :nocov:
+        end
+      end
+
+      # Describes a validation problem with an event.
+      #
+      # @!attribute [r] payload_description
+      #   @return [String] brief description of the part of the event that was invalid
+      # @!attribute [r] message
+      #   @return [String] detailed validation failure message
+      Failure = ::Data.define(:payload_description, :message)
+
+      # Returned by {Interface#validate_event}. Either `failure` is non-nil (the event was invalid)
+      # or `record_preparer` is non-nil (the event was valid and its record can be prepared for
+      # indexing with the given preparer).
+      #
+      # @!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
+        # @implements ValidationResult
+
+        # Builds a result for a valid event.
+        #
+        # @param record_preparer [Object] preparer for the event's record
+        # @return [ValidationResult]
+        def self.valid(record_preparer)
+          new(record_preparer: record_preparer, failure: nil)
+        end
+
+        # Builds a result for an invalid event.
+        #
+        # @param payload_description [String] brief description of the part of the event that was invalid
+        # @param message [String] detailed validation failure message
+        # @return [ValidationResult]
+        def self.invalid(payload_description:, message:)
+          new(record_preparer: nil, failure: Failure.new(payload_description: payload_description, message: message))
+        end
+      end
+    end
+  end
+end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb
new file mode 100644
index 000000000..d6207d19c
--- /dev/null
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb
@@ -0,0 +1,138 @@
+# Copyright 2024 - 2026 Block, Inc.
+#
+# Use of this source code is governed by an MIT-style
+# license that can be found in the LICENSE file or at
+# https://opensource.org/licenses/MIT.
+#
+# frozen_string_literal: true
+
+require "elastic_graph/constants"
+require "elastic_graph/indexer/event_id"
+require "elastic_graph/indexer/ingestion_adapter"
+require "elastic_graph/indexer/record_preparer"
+require "elastic_graph/support/json_schema/validator_factory"
+
+module ElasticGraph
+  class Indexer
+    module IngestionAdapter
+      # Ingestion adapter for events in ElasticGraph's versioned JSON format: it validates events
+      # against the JSON schema identified by the event's `json_schema_version`, and prepares
+      # records using that version's view of the schema.
+      class JSONEvents
+        # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts
+        # @param logger [Logger] the ElasticGraph logger
+        # @param configure_record_validator [Proc, nil] optional callback to further configure the record validator
+        def initialize(schema_artifacts:, logger:, configure_record_validator: nil)
+          @schema_artifacts = schema_artifacts
+          @logger = logger
+          @configure_record_validator = configure_record_validator
+          @record_preparer_factory = RecordPreparer::Factory.new(schema_artifacts)
+        end
+
+        # (see Interface#handles_event?)
+        def handles_event?(event)
+          event.key?(JSON_SCHEMA_VERSION_KEY)
+        end
+
+        # (see Interface#validate_event)
+        def validate_event(event, skip_record_validation: false)
+          selected_json_schema_version = select_json_schema_version(event) { |failure| return failure }
+
+          # Because the `select_json_schema_version` picks the closest-matching json schema version, the incoming
+          # event might not match the expected json_schema_version value in the json schema (which is a `const` field).
+          # This is by design, since we're picking a schema based on best-effort, so to avoid that by-design validation error,
+          # performing the envelope validation on a "patched" version of the event.
+          event_with_patched_envelope = event.merge({JSON_SCHEMA_VERSION_KEY => selected_json_schema_version})
+
+          if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope))
+            return ValidationResult.invalid(payload_description: "event payload", message: error_message)
+          end
+
+          record = event.fetch("record")
+          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(payload_description: "#{graphql_type_name} record", message: error_message)
+          end
+
+          ValidationResult.valid(@record_preparer_factory.for_json_schema_version(selected_json_schema_version))
+        end
+
+        private
+
+        def select_json_schema_version(event)
+          available_json_schema_versions = @schema_artifacts.available_json_schema_versions
+
+          requested_json_schema_version = event[JSON_SCHEMA_VERSION_KEY]
+
+          # First check that a valid value has been requested (a positive integer)
+          if !event.key?(JSON_SCHEMA_VERSION_KEY)
+            yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`")
+          elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1
+            yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_schema_version}) must be a positive integer.")
+          end
+
+          # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema
+          # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available
+          # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match,
+          # the event can still be indexed.
+          #
+          # This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired
+          # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version
+          # should be selected. So to get that behavior, the list is sorted in descending order.
+          #
+          selected_json_schema_version = available_json_schema_versions.sort.reverse.min_by { |version| (requested_json_schema_version - version).abs }
+
+          if selected_json_schema_version != requested_json_schema_version
+            @logger.info({
+              "message_type" => "ElasticGraphMissingJSONSchemaVersion",
+              "message_id" => event["message_id"],
+              "event_id" => EventID.from_event(event),
+              "event_type" => event["type"],
+              "requested_json_schema_version" => requested_json_schema_version,
+              "selected_json_schema_version" => selected_json_schema_version
+            })
+          end
+
+          if selected_json_schema_version.nil?
+            yield ValidationResult.invalid(
+              payload_description: JSON_SCHEMA_VERSION_KEY,
+              message: "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \
+              Available json schema versions: #{available_json_schema_versions.sort.join(", ")}"
+            )
+          end
+
+          selected_json_schema_version
+        end
+
+        def validator(type, selected_json_schema_version)
+          factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory
+          factory.validator_for(type)
+        end
+
+        def validator_factories_by_version
+          @validator_factories_by_version ||= ::Hash.new do |hash, json_schema_version|
+            factory = Support::JSONSchema::ValidatorFactory.new(
+              schema: @schema_artifacts.json_schemas_for(json_schema_version),
+              sanitize_pii: true
+            )
+
+            if (configure_record_validator = @configure_record_validator)
+              factory = configure_record_validator.call(factory)
+            end
+
+            hash[json_schema_version] = factory
+          end
+        end
+
+        # :nocov: -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class.
+        def raise(*args)
+          super("`raise` was called on `IngestionAdapter::JSONEvents`, but should not. Instead, use " \
+            "`yield ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \
+            "the valid events to still be processed.")
+        end
+        # :nocov:
+      end
+    end
+  end
+end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
index 79e36e83d..042afb3d2 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
@@ -6,12 +6,10 @@
 #
 # frozen_string_literal: true
 
-require "elastic_graph/constants"
 require "elastic_graph/indexer/event_id"
 require "elastic_graph/indexer/failed_event_error"
 require "elastic_graph/indexer/operation/update"
 require "elastic_graph/indexer/record_preparer"
-require "elastic_graph/support/json_schema/validator_factory"
 require "elastic_graph/support/memoizable_data"
 require "zlib"
 
@@ -21,42 +19,38 @@ module Operation
       class Factory < Support::MemoizableData.define(
         :schema_artifacts,
         :index_definitions_by_graphql_type,
-        :record_preparer_factory,
+        :ingestion_adapters,
         :logger,
         :skip_derived_indexing_type_updates,
-        :skip_record_validation_percents_by_type,
-        :configure_record_validator
+        :skip_record_validation_percents_by_type
       )
         def build(event)
           event = prepare_event(event)
 
-          selected_json_schema_version = select_json_schema_version(event) { |failure| return failure }
+          unless (adapter = ingestion_adapter_for(event))
+            return build_failed_result(event, "event payload", "No available ingestion adapter recognized this event.")
+          end
 
-          # Because the `select_json_schema_version` picks the closest-matching json schema version, the incoming
-          # event might not match the expected json_schema_version value in the json schema (which is a `const` field).
-          # This is by design, since we're picking a schema based on best-effort, so to avoid that by-design validation error,
-          # performing the envelope validation on a "patched" version of the event.
-          event_with_patched_envelope = event.merge({JSON_SCHEMA_VERSION_KEY => selected_json_schema_version})
+          skip_record_validation = skip_validation?(event["type"], event)
+          validation_result = adapter.validate_event(event, skip_record_validation: skip_record_validation)
 
-          if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope))
-            return build_failed_result(event, "event payload", error_message)
+          if (failure = validation_result.failure)
+            return build_failed_result(event, failure.payload_description, failure.message)
           end
 
-          graphql_type_name = event.fetch("type")
-
-          if skip_validation?(graphql_type_name, event)
-            build_success_result_isolating_malformed_records(event, graphql_type_name, selected_json_schema_version)
+          record_preparer = validation_result.record_preparer # : _RecordPreparer
+          if skip_record_validation
+            build_success_result_isolating_malformed_records(event, record_preparer, adapter)
           else
-            validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version) ||
-              build_success_result(event, selected_json_schema_version, type_with_skipped_validation: nil)
+            build_success_result(event, record_preparer, type_with_skipped_validation: nil)
           end
         end
 
         private
 
-        def build_success_result(event, selected_json_schema_version, type_with_skipped_validation:)
+        def build_success_result(event, record_preparer, type_with_skipped_validation:)
           BuildResult.success(
-            build_all_operations_for(event, record_preparer_factory.for_json_schema_version(selected_json_schema_version)),
+            build_all_operations_for(event, record_preparer),
             type_with_skipped_validation: type_with_skipped_validation
           )
         end
@@ -69,77 +63,23 @@ def build_success_result(event, selected_json_schema_version, type_with_skipped_
         # whether the data was actually bad, and if it was, hands the caller the same pinpointed message it
         # would have gotten had we validated up front. A clean bill of health from the validator means the
         # error was never about the data (a schema artifact defect, or a bug) and must not be swallowed.
-        def build_success_result_isolating_malformed_records(event, graphql_type_name, selected_json_schema_version)
-          build_success_result(event, selected_json_schema_version, type_with_skipped_validation: graphql_type_name)
+        def build_success_result_isolating_malformed_records(event, record_preparer, adapter)
+          build_success_result(event, record_preparer, type_with_skipped_validation: event.fetch("type"))
         rescue => exception
-          failed_result = validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version)
+          failure = adapter.validate_event(event).failure
           # `raise` is overridden below to stop this class from *originating* an error instead of returning a
           # `BuildResult`. Here we propagate one that already escaped a collaborator, which is exactly what
           # happens without this rescue, so we deliberately bypass that guard.
-          ::Kernel.raise(exception) unless failed_result
-          failed_result
+          ::Kernel.raise(exception) unless failure
+          build_failed_result(event, failure.payload_description, failure.message)
         end
 
-        def select_json_schema_version(event)
-          available_json_schema_versions = schema_artifacts.available_json_schema_versions
-
-          requested_json_schema_version = event[JSON_SCHEMA_VERSION_KEY]
-
-          # First check that a valid value has been requested (a positive integer)
-          if !event.key?(JSON_SCHEMA_VERSION_KEY)
-            yield build_failed_result(event, JSON_SCHEMA_VERSION_KEY, "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`")
-          elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1
-            yield build_failed_result(event, JSON_SCHEMA_VERSION_KEY, "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_schema_version}) must be a positive integer.")
-          end
-
-          # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema
-          # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available
-          # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match,
-          # the event can still be indexed.
-          #
-          # This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired
-          # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version
-          # should be selected. So to get that behavior, the list is sorted in descending order.
-          #
-          selected_json_schema_version = available_json_schema_versions.sort.reverse.min_by { |version| (requested_json_schema_version - version).abs }
-
-          if selected_json_schema_version != requested_json_schema_version
-            logger.info({
-              "message_type" => "ElasticGraphMissingJSONSchemaVersion",
-              "message_id" => event["message_id"],
-              "event_id" => EventID.from_event(event),
-              "event_type" => event["type"],
-              "requested_json_schema_version" => requested_json_schema_version,
-              "selected_json_schema_version" => selected_json_schema_version
-            })
-          end
-
-          if selected_json_schema_version.nil?
-            yield build_failed_result(
-              event, JSON_SCHEMA_VERSION_KEY,
-              "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \
-              Available json schema versions: #{available_json_schema_versions.sort.join(", ")}"
-            )
-          end
-
-          selected_json_schema_version
-        end
-
-        def validator(type, selected_json_schema_version)
-          factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory
-          factory.validator_for(type)
-        end
-
-        def validator_factories_by_version
-          @validator_factories_by_version ||= ::Hash.new do |hash, raw_json_schema_version|
-            json_schema_version = raw_json_schema_version # : Integer
-            factory = Support::JSONSchema::ValidatorFactory.new(
-              schema: schema_artifacts.json_schemas_for(json_schema_version),
-              sanitize_pii: true
-            )
-            factory = configure_record_validator.call(factory) if configure_record_validator
-            hash[json_schema_version] = factory
-          end
+        # Routes the event to the first ingestion adapter that recognizes it. When exactly one
+        # adapter is available, it receives all events--including unrecognizable ones--so that
+        # its more specific validation failure messages are used.
+        def ingestion_adapter_for(event)
+          ingestion_adapters.find { |adapter| adapter.handles_event?(event) } ||
+            (ingestion_adapters.first if ingestion_adapters.size == 1)
         end
 
         # This copies the `id` from event into the actual record
@@ -149,15 +89,6 @@ def prepare_event(event)
           event.merge("record" => event["record"].merge("id" => event.fetch("id")))
         end
 
-        def validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version)
-          record = event.fetch("record")
-          validator = validator(graphql_type_name, selected_json_schema_version)
-
-          if (error_message = validator.validate_with_error_message(record))
-            build_failed_result(event, "#{graphql_type_name} record", error_message)
-          end
-        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
@@ -179,9 +110,9 @@ def skip_validation?(type, event)
         def build_failed_result(event, payload_description, validation_message)
           message = "Malformed #{payload_description}. #{validation_message}"
 
-          # Here we use the `RecordPreparer::Identity` record preparer because we may not have a valid JSON schema
-          # version number in this case (which is usually required to get a `RecordPreparer` from the factory), and
-          # we won't wind up using the record preparer for real on these operations, anyway.
+          # Here we use the `RecordPreparer::Identity` record preparer because the event failed validation, so
+          # no adapter-provided record preparer is available, and we won't wind up using the record preparer
+          # for real on these operations, anyway.
           #
           # Building operations for an event we already know is malformed can itself fail--for example, when the
           # record omits a field an update target derives its id from. Reporting what was malformed matters more
@@ -201,6 +132,7 @@ def build_failed_result(event, payload_description, validation_message)
             [] # : ::Array[operation]
           end
 
+
           BuildResult.failure(FailedEventError.new(event: event, operations: operations.to_set, main_message: message))
         end
 
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb
index bdfc2791c..d1a705e59 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb
@@ -6,14 +6,21 @@
 #
 # frozen_string_literal: true
 
+require "elastic_graph/indexer/ingestion_adapter/json_events"
 require "json"
 
 # Defines an RSpec matcher that can be used to validate ElasticGraph events.
 ::RSpec::Matchers.define :be_a_valid_elastic_graph_event do |for_indexer:|
   match do |event|
+    json_events_adapter = ElasticGraph::Indexer::IngestionAdapter::JSONEvents.new(
+      schema_artifacts: for_indexer.schema_artifacts,
+      logger: for_indexer.logger,
+      configure_record_validator: block_arg
+    )
+
     result = for_indexer
       .operation_factory
-      .with(configure_record_validator: block_arg)
+      .with(ingestion_adapters: [json_events_adapter])
       .build(event)
 
     @validation_failure = result.failed_event_error
diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs
index 6f1d1bc40..c876f01d2 100644
--- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs
+++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs
@@ -28,6 +28,9 @@ module ElasticGraph
     @operation_factory: Operation::Factory?
     def operation_factory: () -> Operation::Factory
 
+    @ingestion_adapters: ::Array[IngestionAdapter::_Adapter]?
+    def ingestion_adapters: () -> ::Array[IngestionAdapter::_Adapter]
+
     @monotonic_clock: Support::MonotonicClock?
     def monotonic_clock: () -> Support::MonotonicClock
 
diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs
new file mode 100644
index 000000000..47a17ee56
--- /dev/null
+++ b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs
@@ -0,0 +1,42 @@
+module ElasticGraph
+  class Indexer
+    module IngestionAdapter
+      interface _Adapter
+        def handles_event?: (event) -> bool
+        def validate_event: (event, ?skip_record_validation: bool) -> ValidationResult
+      end
+
+      class Interface
+        include _Adapter
+
+        def initialize: (schema_artifacts: schemaArtifacts, logger: ::Logger) -> void
+      end
+
+      class FailureSupertype < ::Data
+        attr_reader payload_description: ::String
+        attr_reader message: ::String
+
+        def initialize: (payload_description: ::String, message: ::String) -> void
+        def self.new: (payload_description: ::String, message: ::String) -> instance
+        def self.members: () -> ::Array[::Symbol]
+      end
+
+      class Failure < FailureSupertype
+      end
+
+      class ValidationResultSupertype < ::Data
+        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 self.members: () -> ::Array[::Symbol]
+      end
+
+      class ValidationResult < ValidationResultSupertype
+        def self.valid: (_RecordPreparer) -> ValidationResult
+        def self.invalid: (payload_description: ::String, message: ::String) -> ValidationResult
+      end
+    end
+  end
+end
diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs
new file mode 100644
index 000000000..7322249f9
--- /dev/null
+++ b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs
@@ -0,0 +1,31 @@
+module ElasticGraph
+  class Indexer
+    module IngestionAdapter
+      class JSONEvents
+        def initialize: (
+          schema_artifacts: schemaArtifacts,
+          logger: ::Logger,
+          ?configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)?
+        ) -> void
+
+        @schema_artifacts: schemaArtifacts
+        @logger: ::Logger
+        @configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)?
+        @record_preparer_factory: RecordPreparer::Factory
+
+        def handles_event?: (event) -> bool
+        def validate_event: (event, ?skip_record_validation: bool) -> ValidationResult
+
+        private
+
+        def select_json_schema_version: (event) { (ValidationResult) -> bot } -> (::Integer | bot)
+        def validator: (::String, ::Integer) -> Support::JSONSchema::Validator
+
+        @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]?
+        def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]
+
+        def raise: (*untyped) -> bot
+      end
+    end
+  end
+end
diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs
index a07dfb922..a3ac77b44 100644
--- a/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs
+++ b/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs
@@ -1,37 +1,32 @@
 module ElasticGraph
   class Indexer
     type event = ::Hash[::String, untyped]
-    type validator = Support::JSONSchema::Validator
-    type validatorFactory = Support::JSONSchema::ValidatorFactory
 
     module Operation
       class FactorySupertype
         attr_reader schema_artifacts: schemaArtifacts
         attr_reader index_definitions_by_graphql_type: ::Hash[::String, ::Array[DatastoreCore::_IndexDefinition]]
-        attr_reader record_preparer_factory: RecordPreparer::Factory
+        attr_reader ingestion_adapters: ::Array[IngestionAdapter::_Adapter]
         attr_reader logger: ::Logger
         attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]
         attr_reader skip_record_validation_percents_by_type: ::Hash[::String, ::Float]
-        attr_reader configure_record_validator: (^(validatorFactory) -> validatorFactory)?
 
         def initialize: (
           schema_artifacts: schemaArtifacts,
           index_definitions_by_graphql_type: ::Hash[::String, ::Array[DatastoreCore::_IndexDefinition]],
-          record_preparer_factory: RecordPreparer::Factory,
+          ingestion_adapters: ::Array[IngestionAdapter::_Adapter],
           logger: ::Logger,
           skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
-          skip_record_validation_percents_by_type: ::Hash[::String, ::Float],
-          configure_record_validator: (^(validatorFactory) -> validatorFactory)?
+          skip_record_validation_percents_by_type: ::Hash[::String, ::Float]
         ) -> void
 
         def with: (
           ?schema_artifacts: schemaArtifacts,
           ?index_definitions_by_graphql_type: ::Hash[::String, ::Array[DatastoreCore::_IndexDefinition]],
-          ?record_preparer_factory: RecordPreparer::Factory,
+          ?ingestion_adapters: ::Array[IngestionAdapter::_Adapter],
           ?logger: ::Logger,
           ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
-          ?skip_record_validation_percents_by_type: ::Hash[::String, ::Float],
-          ?configure_record_validator: (^(validatorFactory) -> validatorFactory)?
+          ?skip_record_validation_percents_by_type: ::Hash[::String, ::Float]
         ) -> instance
       end
 
@@ -40,16 +35,10 @@ module ElasticGraph
 
         private
 
-        def validator: (::String, ::Integer) -> Support::JSONSchema::Validator
-
-        @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]?
-        def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]
-
-        def select_json_schema_version: (event) { (BuildResult) -> bot } -> (::Integer | bot)
+        def ingestion_adapter_for: (event) -> IngestionAdapter::_Adapter?
         def prepare_event: (event) -> event
-        def validate_record_returning_failure: (event, ::String, ::Integer) -> BuildResult?
-        def build_success_result: (event, ::Integer, type_with_skipped_validation: ::String?) -> BuildResult
-        def build_success_result_isolating_malformed_records: (event, ::String, ::Integer) -> BuildResult
+        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
         def skip_validation?: (::String, event) -> bool
         def build_failed_result: (event, ::String, ::String) -> BuildResult
diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
index e2162f611..c7c739db4 100644
--- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
+++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
@@ -6,8 +6,9 @@
 #
 # frozen_string_literal: true
 
-require "elastic_graph/indexer"
 require "elastic_graph/constants"
+require "elastic_graph/indexer"
+require "elastic_graph/indexer/ingestion_adapter/json_events"
 require "elastic_graph/indexer/operation/factory"
 require "elastic_graph/spec_support/builds_indexer_operation"
 require "json"
@@ -311,12 +312,17 @@ module Operation
           # A factory whose record preparation fails for a reason record validation cannot detect: a
           # runtime metadata defect, standing in for any bug that is not about the data itself.
           def factory_whose_record_preparation_is_broken
-            record_preparer_factory = instance_double(RecordPreparer::Factory)
+            record_preparer = instance_double(RecordPreparer)
 
-            allow(record_preparer_factory).to receive(:for_json_schema_version)
+            allow(record_preparer).to receive(:prepare_for_index)
               .and_raise(::KeyError, 'key not found: "nameInIndex"')
 
-            indexer.operation_factory.with(record_preparer_factory: record_preparer_factory)
+            adapter = indexer.ingestion_adapters.first
+            allow(adapter).to receive(:validate_event).and_wrap_original do |original, *args, **kwargs|
+              original.call(*args, **kwargs).with(record_preparer: record_preparer)
+            end
+
+            indexer.operation_factory
           end
 
           it "generates a primary indexing operation for a single index with latency metrics" do
@@ -677,6 +683,58 @@ def factory_whose_record_preparation_is_broken
             end
           end
 
+          context "when multiple ingestion adapters are available" do
+            it "routes each event to the first adapter that recognizes it" do
+              event = build_upsert_event(:component, id: "1", __version: 1)
+
+              non_matching_adapter = instance_double(IngestionAdapter::Interface, handles_event?: false)
+              matching_adapter = instance_double(
+                IngestionAdapter::Interface,
+                handles_event?: true,
+                validate_event: IngestionAdapter::ValidationResult.valid(RecordPreparer::Identity)
+              )
+
+              factory = indexer.operation_factory.with(ingestion_adapters: [non_matching_adapter, matching_adapter])
+              result = factory.build(event)
+
+              expect(result.failed_event_error).to be nil
+              expect(result.operations).not_to be_empty
+              expect(matching_adapter).to have_received(:validate_event).with(a_hash_including("type" => "Component"), skip_record_validation: false)
+            end
+
+            it "fails the event when no adapter recognizes it" do
+              event = build_upsert_event(:component, id: "1", __version: 1)
+
+              adapters = [
+                instance_double(IngestionAdapter::Interface, handles_event?: false),
+                instance_double(IngestionAdapter::Interface, handles_event?: false)
+              ]
+
+              factory = indexer.operation_factory.with(ingestion_adapters: adapters)
+
+              expect_failed_event_error(event, "No available ingestion adapter recognized this event.", factory: factory)
+            end
+          end
+
+          context "when a single ingestion adapter is available" do
+            it "routes all events to it, even ones it does not recognize, so that its more specific failure messages are used" do
+              event = build_upsert_event(:component, id: "1", __version: 1)
+
+              adapter = instance_double(
+                IngestionAdapter::Interface,
+                handles_event?: false,
+                validate_event: IngestionAdapter::ValidationResult.invalid(
+                  payload_description: "event payload",
+                  message: "not recognizable by this adapter"
+                )
+              )
+
+              factory = indexer.operation_factory.with(ingestion_adapters: [adapter])
+
+              expect_failed_event_error(event, "not recognizable by this adapter", factory: factory)
+            end
+          end
+
           def expect_failed_event_error(event, *error_message_snippets, factory: indexer.operation_factory, expect_no_ops: false)
             result = factory.build(event)
 
@@ -718,10 +776,19 @@ def event_id_from(event)
         end
 
         def build_expecting_success(event, **options, &configure_record_validator)
-          result = indexer
-            .operation_factory
-            .with(configure_record_validator: configure_record_validator)
-            .build(event, **options)
+          factory = indexer.operation_factory
+
+          if configure_record_validator
+            json_events_adapter = IngestionAdapter::JSONEvents.new(
+              schema_artifacts: indexer.schema_artifacts,
+              logger: indexer.logger,
+              configure_record_validator: configure_record_validator
+            )
+
+            factory = factory.with(ingestion_adapters: [json_events_adapter])
+          end
+
+          result = factory.build(event, **options)
 
           raise result.failed_event_error if result.failed_event_error
           result.operations

From 3e376cef36e7ccbeab29ef2280416d973891bc32 Mon Sep 17 00:00:00 2001
From: Josh Wilson <joshuaw@squareup.com>
Date: Sat, 5 Sep 2026 14:07:33 -0500
Subject: [PATCH 2/3] Polish ingestion adapter Ruby conventions

---
 .../lib/elastic_graph/indexer/ingestion_adapter.rb        | 8 ++++----
 .../lib/elastic_graph/indexer/operation/factory.rb        | 7 ++++---
 2 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
index e9a4d1c1d..5e5975a6d 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb
@@ -28,9 +28,9 @@ def initialize(schema_artifacts:, logger:)
         # @param event [Hash<String, Object>] an ElasticGraph indexing event
         # @return [Boolean] whether this adapter handles the event
         def handles_event?(event)
-          # :nocov: -- must return a boolean to satisfy Steep type checking but never called
+          # simplecov:disable -- must return a boolean to satisfy Steep type checking but never called
           false
-          # :nocov:
+          # simplecov:enable
         end
 
         # Validates the given event and resolves the record preparer appropriate for the event's
@@ -40,9 +40,9 @@ def handles_event?(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)
-          # :nocov: -- must return a result to satisfy Steep type checking but never called
+          # simplecov:disable -- must return a result to satisfy Steep type checking but never called
           ValidationResult.valid(RecordPreparer::Identity)
-          # :nocov:
+          # simplecov:enable
         end
       end
 
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
index 042afb3d2..66bb0ceae 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
@@ -78,8 +78,9 @@ def build_success_result_isolating_malformed_records(event, record_preparer, ada
         # adapter is available, it receives all events--including unrecognizable ones--so that
         # its more specific validation failure messages are used.
         def ingestion_adapter_for(event)
-          ingestion_adapters.find { |adapter| adapter.handles_event?(event) } ||
-            (ingestion_adapters.first if ingestion_adapters.size == 1)
+          return ingestion_adapters.first if ingestion_adapters.one?
+
+          ingestion_adapters.find { |adapter| adapter.handles_event?(event) }
         end
 
         # This copies the `id` from event into the actual record
@@ -182,7 +183,7 @@ def index_definitions_for(type)
         # simplecov:disable -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class.
         def raise(*args)
           super("`raise` was called on `Operation::Factory`, but should not. Instead, use " \
-            "`yield build_failed_result(...)` so that we can accumulate all invalid events and allow " \
+            "`return build_failed_result(...)` so that we can accumulate all invalid events and allow " \
             "the valid events to still be processed.")
         end
         # simplecov:enable

From 71bcc701b7c919efa0308ec617a564f471ad0322 Mon Sep 17 00:00:00 2001
From: Josh Wilson <joshuaw@squareup.com>
Date: Sat, 5 Sep 2026 14:12:22 -0500
Subject: [PATCH 3/3] Preserve validator typing and dispatch coverage

---
 .../elastic_graph/indexer/ingestion_adapter/json_events.rb | 7 ++++---
 .../lib/elastic_graph/indexer/operation/factory.rb         | 1 -
 .../unit/elastic_graph/indexer/operation/factory_spec.rb   | 7 ++-----
 3 files changed, 6 insertions(+), 9 deletions(-)

diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb
index d6207d19c..e8970c375 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb
@@ -111,7 +111,8 @@ def validator(type, selected_json_schema_version)
         end
 
         def validator_factories_by_version
-          @validator_factories_by_version ||= ::Hash.new do |hash, json_schema_version|
+          @validator_factories_by_version ||= ::Hash.new do |hash, raw_json_schema_version|
+            json_schema_version = raw_json_schema_version # : Integer
             factory = Support::JSONSchema::ValidatorFactory.new(
               schema: @schema_artifacts.json_schemas_for(json_schema_version),
               sanitize_pii: true
@@ -125,13 +126,13 @@ def validator_factories_by_version
           end
         end
 
-        # :nocov: -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class.
+        # simplecov:disable -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class.
         def raise(*args)
           super("`raise` was called on `IngestionAdapter::JSONEvents`, but should not. Instead, use " \
             "`yield ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \
             "the valid events to still be processed.")
         end
-        # :nocov:
+        # simplecov:enable
       end
     end
   end
diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
index 66bb0ceae..755aeb457 100644
--- a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
+++ b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb
@@ -133,7 +133,6 @@ def build_failed_result(event, payload_description, validation_message)
             [] # : ::Array[operation]
           end
 
-
           BuildResult.failure(FailedEventError.new(event: event, operations: operations.to_set, main_message: message))
         end
 
diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
index c7c739db4..21d96ba76 100644
--- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
+++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb
@@ -688,11 +688,8 @@ def factory_whose_record_preparation_is_broken
               event = build_upsert_event(:component, id: "1", __version: 1)
 
               non_matching_adapter = instance_double(IngestionAdapter::Interface, handles_event?: false)
-              matching_adapter = instance_double(
-                IngestionAdapter::Interface,
-                handles_event?: true,
-                validate_event: IngestionAdapter::ValidationResult.valid(RecordPreparer::Identity)
-              )
+              matching_adapter = indexer.ingestion_adapters.first
+              allow(matching_adapter).to receive(:validate_event).and_call_original
 
               factory = indexer.operation_factory.with(ingestion_adapters: [non_matching_adapter, matching_adapter])
               result = factory.build(event)