Move JSON ingestion into elasticgraph-json_ingestion via an indexer extension

Steepfile

@@ -41,9 +41,9 @@ target :elasticgraph_gems do
     elasticgraph-graphql/lib/elastic_graph/graphql/schema/type.rb
   ])
 
-  # elasticgraph-indexer: existing files that don't type check yet.
+  # elasticgraph-json_ingestion: RSpec matcher DSL that doesn't type check.
   ignore(*%w[
-    elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb
+    elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/spec_support/event_matcher.rb
   ])
 
   # elasticgraph-local: existing files that don't type check yet.

config/schema/artifacts/runtime_metadata.yaml

@@ -3219,6 +3219,10 @@ index_definitions_by_name:
       frequency: yearly
       timestamp_field_path: created_at
     route_with: workspace_id2
+indexer_extension_modules:
+- extension_ref:
+    name: ElasticGraph::JSONIngestion::IndexerExtension
+    require_path: elastic_graph/json_ingestion/indexer_extension
 object_types_by_name:
   Address:
     graphql_fields_by_name:

config/schema/artifacts_with_apollo/runtime_metadata.yaml

@@ -3248,6 +3248,10 @@ index_definitions_by_name:
       frequency: yearly
       timestamp_field_path: created_at
     route_with: workspace_id2
+indexer_extension_modules:
+- extension_ref:
+    name: ElasticGraph::JSONIngestion::IndexerExtension
+    require_path: elastic_graph/json_ingestion/indexer_extension
 object_types_by_name:
   Address:
     graphql_fields_by_name:

elasticgraph-apollo/elasticgraph-apollo.gemspec

@@ -52,6 +52,9 @@ Gem::Specification.new do |spec|
   spec.add_development_dependency "elasticgraph-schema_definition", ElasticGraph::VERSION
   spec.add_development_dependency "elasticgraph-admin", ElasticGraph::VERSION
   spec.add_development_dependency "elasticgraph-elasticsearch", ElasticGraph::VERSION
+  # The test suite builds indexers from the shared test schema artifacts, whose runtime metadata
+  # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`.
+  spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION
   spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION
   spec.add_development_dependency "elasticgraph-indexer", ElasticGraph::VERSION
 end

elasticgraph-indexer/lib/elastic_graph/indexer.rb

@@ -7,6 +7,7 @@
 # frozen_string_literal: true
 
 require "elastic_graph/datastore_core"
+require "elastic_graph/errors"
 require "elastic_graph/indexer/config"
 require "elastic_graph/support/from_yaml_file"
 
@@ -56,21 +57,11 @@ def datastore_router
       end
     end
 
-    def record_preparer_factory
-      @record_preparer_factory ||= begin
-        require "elastic_graph/indexer/record_preparer"
-        RecordPreparer::Factory.new(schema_artifacts)
-      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.
+    # The ingestion adapters available for processing events. Returns an empty list by default;
+    # ingestion format gems contribute adapters via indexer extension modules that override this
+    # method.
     def ingestion_adapters
-      @ingestion_adapters ||= begin
-        require "elastic_graph/indexer/ingestion_adapter/json_events"
-        [IngestionAdapter::JSONEvents.new(schema_artifacts: schema_artifacts, logger: logger)]
-      end
+      [] # : ::Array[IngestionAdapter::_Adapter]
     end
 
     def processor
@@ -88,6 +79,13 @@ def processor
 
     def operation_factory
       @operation_factory ||= begin
+        if ingestion_adapters.empty?
+          raise Errors::ConfigError, "No ingestion adapters are available to process events. Ingestion format gems " \
+            "make an adapter available by registering an indexer extension (via `register_indexer_extension`) during " \
+            "schema definition; ensure your schema definition uses an ingestion format extension and regenerate your " \
+            "schema artifacts (or configure an extension via the `indexer.extension_modules` setting)."
+        end
+
         require "elastic_graph/indexer/operation/factory"
         Operation::Factory.new(
           schema_artifacts: schema_artifacts,

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, ingestion_adapters, logger
+    # @dynamic processor, operation_factory, ingestion_adapters, logger
     # @dynamic self.from_parsed_yaml
   end
 end

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

@@ -11,58 +11,6 @@
 module ElasticGraph
   class Indexer
     class RecordPreparer
-      # Provides the ability to get a `RecordPreparer` for a specific JSON schema version.
-      class Factory
-        def initialize(schema_artifacts)
-          @schema_artifacts = schema_artifacts
-
-          scalar_types_by_name = schema_artifacts.runtime_metadata.scalar_types_by_name
-          indexing_preparer_by_scalar_type_name = ::Hash.new do |hash, type_name|
-            hash[type_name] = scalar_types_by_name[type_name]&.load_indexing_preparer&.extension_class
-          end # : ::Hash[::String, SchemaArtifacts::RuntimeMetadata::extensionClass?]
-
-          @preparers_by_json_schema_version = ::Hash.new do |hash, version|
-            hash[version] = RecordPreparer.new(
-              indexing_preparer_by_scalar_type_name,
-              build_type_metas_from(@schema_artifacts.json_schemas_for(version))
-            )
-          end
-        end
-
-        # Gets the `RecordPreparer` for the given JSON schema version.
-        def for_json_schema_version(json_schema_version)
-          @preparers_by_json_schema_version[json_schema_version] # : RecordPreparer
-        end
-
-        # Gets the `RecordPreparer` for the latest JSON schema version. Intended primarily
-        # for use in tests for convenience.
-        def for_latest_json_schema_version
-          for_json_schema_version(@schema_artifacts.latest_json_schema_version)
-        end
-
-        private
-
-        def build_type_metas_from(json_schemas)
-          json_schemas.fetch("$defs").filter_map do |type, type_def|
-            next if type == EVENT_ENVELOPE_JSON_SCHEMA_NAME
-
-            properties = type_def.fetch("properties") do
-              {} # : ::Hash[::String, untyped]
-            end # : ::Hash[::String, untyped]
-
-            eg_meta_by_field_name = properties.filter_map do |prop_name, prop|
-              eg_meta = prop["ElasticGraph"]
-              [prop_name, eg_meta] if eg_meta
-            end.to_h
-
-            TypeMetadata.new(
-              name: type,
-              eg_meta_by_field_name: eg_meta_by_field_name
-            )
-          end
-        end
-      end
-
       # An alternate `RecordPreparer` implementation that implements the identity function:
       # it just echoes back the record it is given.
       #

elasticgraph-indexer/sig/elastic_graph/indexer.rbs

@@ -19,18 +19,14 @@ module ElasticGraph
     @datastore_router: Indexer::_DatastoreRouter?
     def datastore_router: () -> Indexer::_DatastoreRouter
 
-    @record_preparer_factory: RecordPreparer::Factory?
-    def record_preparer_factory: () -> RecordPreparer::Factory
+    def ingestion_adapters: () -> ::Array[IngestionAdapter::_Adapter]
 
     @processor: Processor?
     def processor: () -> Processor
 
     @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
 

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

@@ -8,20 +8,6 @@ module ElasticGraph
       type egMetaByFieldHash = ::Hash[::String, {"type" => ::String, "nameInIndex" => ::String}]
       type egMetaByFieldByTypeHash = ::Hash[::String, egMetaByFieldHash]
 
-      class Factory
-        def initialize: (schemaArtifacts) -> void
-
-        def for_json_schema_version: (::Integer) -> RecordPreparer
-        def for_latest_json_schema_version: () -> RecordPreparer
-
-        private
-
-        @schema_artifacts: schemaArtifacts
-        @preparers_by_json_schema_version: ::Hash[::Integer, RecordPreparer]
-
-        def build_type_metas_from: (::Hash[::String, untyped]) -> ::Array[TypeMetadata]
-      end
-
       include _RecordPreparer
 
       module Identity

elasticgraph-indexer/spec/support/indexing_preparer.rb

@@ -29,7 +29,7 @@
       schema.object_type "Object" do |t|
         t.field "scalar", scalar_type
       end
-    end).record_preparer_factory.for_latest_json_schema_version
+    end).then { |indexer| latest_json_record_preparer_for(indexer) }
   end
 
   def prepare_scalar_value(value)

elasticgraph-indexer/spec/unit/elastic_graph/indexer/datastore_indexing_router_spec.rb

@@ -601,7 +601,7 @@ def new_operation(event, update_target: nil, **overrides)
 
         arguments = {
           event: event,
-          prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index(
+          prepared_record: latest_json_record_preparer_for(indexer).prepare_for_index(
             event.fetch("type"),
             event.fetch("record"),
             destination_index_mapping.fetch("properties")

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

@@ -300,7 +300,7 @@ def script_params_for(data:, source_type:, destination_type:, indexer: build_ind
           update = Update.new(
             event: {"type" => source_type, "record" => data},
             destination_index_def: destination_index_def,
-            prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index(
+            prepared_record: latest_json_record_preparer_for(indexer).prepare_for_index(
               source_type,
               data,
               destination_index_mapping.fetch("properties")