Move JSON ingestion into elasticgraph-json_ingestion via an indexer extension

The JSON ingestion adapter and JSON-schema-based record preparer factory
move from elasticgraph-indexer to elasticgraph-json_ingestion, and the
JSON ingestion schema definition extension registers an indexer extension
in runtime metadata that makes the adapter available at indexing time.
Any schema defined with JSON ingestion support automatically gets JSON
event ingestion--no settings needed--and elasticgraph-indexer no longer
contains (or refers to) any JSON-specific ingestion logic.

The be_a_valid_elastic_graph_event matcher moves to
elastic_graph/json_ingestion/spec_support/event_matcher, and suites that
build indexers from the shared test schema artifacts gain a development
dependency on elasticgraph-json_ingestion (whose indexer extension those
artifacts now reference).

Author: jwils

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/ingestion_adapter/json_events.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/lib/elastic_graph/indexer/record_preparer.rb

@@ -19,16 +19,12 @@ module ElasticGraph
     @datastore_router: Indexer::_DatastoreRouter?
     def datastore_router: () -> Indexer::_DatastoreRouter
 
-    @record_preparer_factory: RecordPreparer::Factory?
-    def record_preparer_factory: () -> RecordPreparer::Factory
-
     @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?