Generalize indexing schema versions

elasticgraph-indexer/README.md

@@ -48,3 +48,33 @@ indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml")
 events = [] # JSON events read from an async datastream
 indexer.processor.process(events)
 ```
+
+## Custom Payload Decoding
+
+`ElasticGraph::Indexer` can be configured with an indexing event decoder extension. Decoders turn raw payload strings
+from a transport into ElasticGraph indexing event hashes before the normal validation and indexing pipeline runs. The
+default decoder expects JSON Lines.
+
+```yaml
+indexer:
+  indexing_event_decoder:
+    name: MyCompany::ElasticGraph::CSVIndexingEventDecoder
+    require_path: ./lib/my_company/elastic_graph/csv_indexing_event_decoder
+    config:
+      delimiter: ","
+```
+
+Decoder extensions must implement:
+
+```ruby
+def initialize(config:, schema_artifacts:, logger:)
+end
+
+def decode(payload)
+  # return an array of ElasticGraph indexing event hashes
+end
+```
+
+Decoded event hashes do not need to provide a schema version. When a version is omitted, the latest
+available schema artifact version is used for validation and record preparation. Decoders may include
+`schema_version` to request a specific schema artifact version.

elasticgraph-indexer/lib/elastic_graph/indexer.rb

@@ -86,6 +86,18 @@ def operation_factory
       end
     end
 
+    def indexing_event_decoder
+      @indexing_event_decoder ||= begin
+        extension = config.indexing_event_decoder
+        decoder_class = extension.extension_class # : untyped
+        decoder_class.new(
+          config: extension.config,
+          schema_artifacts: schema_artifacts,
+          logger: logger
+        )
+      end
+    end
+
     def monotonic_clock
       @monotonic_clock ||= begin
         require "elastic_graph/support/monotonic_clock"

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

@@ -6,12 +6,19 @@
 #
 # frozen_string_literal: true
 
-require "elastic_graph/support/config"
 require "elastic_graph/errors"
+require "elastic_graph/indexer/indexing_event_decoder"
+require "elastic_graph/schema_artifacts/runtime_metadata/extension_loader"
+require "elastic_graph/support/config"
 
 module ElasticGraph
   class Indexer
-    class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates)
+    class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :indexing_event_decoder)
+      DEFAULT_INDEXING_EVENT_DECODER = {
+        "name" => "ElasticGraph::Indexer::IndexingEventDecoder::JSONLines",
+        "require_path" => "elastic_graph/indexer/indexing_event_decoder"
+      }
+
       json_schema at: "indexer",
         optional: false,
         description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.",
@@ -42,17 +49,64 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms
               {}, # : untyped
               {"WidgetWorkspace" => ["ABC12345678"]}
             ]
+          },
+          indexing_event_decoder: {
+            description: "Extension object used to decode raw indexing payloads into ElasticGraph indexing event hashes. The default decoder expects JSON Lines.",
+            type: "object",
+            properties: {
+              name: {
+                description: "The name of the indexing event decoder extension class.",
+                type: "string",
+                pattern: /^[A-Z]\w+(::[A-Z]\w+)*$/.source, # https://rubular.com/r/UuqAz4fR3kdMip
+                examples: ["MyCompany::ElasticGraph::CSVIndexingEventDecoder"]
+              },
+              require_path: {
+                description: "The path to require to load the indexing event decoder extension.",
+                type: "string",
+                minLength: 1,
+                examples: ["./lib/my_company/elastic_graph/csv_indexing_event_decoder"]
+              },
+              config: {
+                description: "Configuration for the indexing event decoder. Will be passed into the decoder's `#initialize` method.",
+                type: "object",
+                default: {}, # : untyped
+                examples: [
+                  {}, # : untyped
+                  {"delimiter" => ","}
+                ]
+              }
+            },
+            required: ["name", "require_path"],
+            default: DEFAULT_INDEXING_EVENT_DECODER,
+            examples: [
+              DEFAULT_INDEXING_EVENT_DECODER,
+              {
+                "name" => "MyCompany::ElasticGraph::CSVIndexingEventDecoder",
+                "require_path" => "./lib/my_company/elastic_graph/csv_indexing_event_decoder",
+                "config" => {"delimiter" => ","}
+              }
+            ]
           }
         }
 
       private
 
-      def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:)
+      def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, indexing_event_decoder:)
         {
           skip_derived_indexing_type_updates: skip_derived_indexing_type_updates.transform_values(&:to_set),
-          latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms
+          latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms,
+          indexing_event_decoder: load_indexing_event_decoder(indexing_event_decoder)
         }
       end
+
+      def load_indexing_event_decoder(config)
+        loader = SchemaArtifacts::RuntimeMetadata::ExtensionLoader.new(IndexingEventDecoder::Interface)
+        loader.load(
+          config.fetch("name"),
+          from: config.fetch("require_path"),
+          config: config["config"] || {}
+        )
+      end
     end
   end
 end

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

@@ -12,7 +12,7 @@ module ElasticGraph
   class Indexer
     # A unique identifier for an event ingested by the indexer. As a string, takes the form of
     # "[type]:[id]@v[version]", such as "Widget:123abc@v7". This format was designed to make it
-    # easy to put these ids in a comma-seperated list.
+    # easy to put these ids in a comma-separated list.
     EventID = ::Data.define(:type, :id, :version) do
       # @implements EventID
       def self.from_event(event)
@@ -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, indexing_event_decoder, logger
     # @dynamic self.from_parsed_yaml
   end
 end

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

@@ -0,0 +1,50 @@
+# 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 "json"
+
+module ElasticGraph
+  class Indexer
+    # Namespace for indexing event decoders, which turn raw payload strings from a transport into
+    # ElasticGraph indexing event hashes. The decoder to use is configured via the
+    # `indexer.indexing_event_decoder` setting.
+    module IndexingEventDecoder
+      # Defines the indexing event decoder interface, which our extension loader will validate against.
+      class Interface
+        # @param config [Hash<String, Object>] configuration from the `indexing_event_decoder.config` setting
+        # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts
+        # @param logger [Logger] the ElasticGraph logger
+        def initialize(config:, schema_artifacts:, logger:)
+          # must be defined, but nothing to do
+        end
+
+        # @param payload [String] a raw payload from the transport
+        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events. Events do not
+        #   need to include a schema version; when omitted, the latest available schema version is used.
+        def decode(payload)
+          # :nocov: -- must return an array to satisfy Steep type checking but never called
+          []
+          # :nocov:
+        end
+      end
+
+      # The default indexing event decoder, which expects newline-delimited JSON objects.
+      class JSONLines < Interface
+        # (see Interface#initialize)
+        def initialize(config:, schema_artifacts:, logger:)
+          # must be defined for extension interface verification, but nothing to do
+        end
+
+        # (see Interface#decode)
+        def decode(payload)
+          payload.split("\n").map { |event| JSON.parse(event) }
+        end
+      end
+    end
+  end
+end

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

@@ -28,37 +28,30 @@ class Factory < Support::MemoizableData.define(
         def build(event)
           event = prepare_event(event)
 
-          selected_json_schema_version = select_json_schema_version(event) { |failure| return failure }
+          requested_schema_version = schema_version_from(event)
+          selected_schema_version = select_schema_version(event, requested_schema_version) { |failure| return failure }
+          event = event.merge(SCHEMA_VERSION_KEY => requested_schema_version)
 
-          # 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})
+          event_for_validation = schema_artifacts.event_for_schema_version_validation(event, selected_schema_version)
 
-          if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope))
+          if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_schema_version).validate_with_error_message(event_for_validation))
             return build_failed_result(event, "event payload", error_message)
           end
 
-          failed_result = validate_record_returning_failure(event, selected_json_schema_version)
+          failed_result = validate_record_returning_failure(event, selected_schema_version)
           failed_result || BuildResult.success(build_all_operations_for(
             event,
-            record_preparer_factory.for_json_schema_version(selected_json_schema_version)
+            record_preparer_factory.for_schema_version(selected_schema_version)
           ))
         end
 
         private
 
-        def select_json_schema_version(event)
-          available_json_schema_versions = schema_artifacts.available_json_schema_versions
+        def select_schema_version(event, requested_schema_version)
+          available_schema_versions = schema_artifacts.available_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.")
+          unless requested_schema_version.is_a?(Integer) && requested_schema_version >= 1
+            yield build_failed_result(event, SCHEMA_VERSION_KEY, "#{SCHEMA_VERSION_KEY} (#{requested_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
@@ -67,46 +60,46 @@ def select_json_schema_version(event)
           # 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
+          # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available 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 }
+          selected_schema_version = available_schema_versions.sort.reverse.min_by { |version| (requested_schema_version - version).abs }
 
-          if selected_json_schema_version != requested_json_schema_version
+          if selected_schema_version != requested_schema_version
             logger.info({
-              "message_type" => "ElasticGraphMissingJSONSchemaVersion",
+              "message_type" => "ElasticGraphMissingSchemaVersion",
               "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
+              "requested_schema_version" => requested_schema_version,
+              "selected_schema_version" => selected_schema_version
             })
           end
 
-          if selected_json_schema_version.nil?
+          if selected_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(", ")}"
+              event, SCHEMA_VERSION_KEY,
+              "Failed to select schema version. Requested version: #{requested_schema_version}. \
+              Available schema versions: #{available_schema_versions.sort.join(", ")}"
             )
           end
 
-          selected_json_schema_version
+          selected_schema_version
         end
 
-        def validator(type, selected_json_schema_version)
-          factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory
+        def validator(type, selected_schema_version)
+          factory = validator_factories_by_version[selected_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|
+          @validator_factories_by_version ||= ::Hash.new do |hash, schema_version|
             factory = Support::JSONSchema::ValidatorFactory.new(
-              schema: schema_artifacts.json_schemas_for(json_schema_version),
+              schema: schema_artifacts.json_schemas_for(schema_version),
               sanitize_pii: true
             )
             factory = configure_record_validator.call(factory) if configure_record_validator
-            hash[json_schema_version] = factory
+            hash[schema_version] = factory
           end
         end
 
@@ -117,10 +110,14 @@ def prepare_event(event)
           event.merge("record" => event["record"].merge("id" => event.fetch("id")))
         end
 
-        def validate_record_returning_failure(event, selected_json_schema_version)
+        def schema_version_from(event)
+          event.fetch(SCHEMA_VERSION_KEY) { schema_artifacts.available_schema_versions.max }
+        end
+
+        def validate_record_returning_failure(event, selected_schema_version)
           record = event.fetch("record")
           graphql_type_name = event.fetch("type")
-          validator = validator(graphql_type_name, selected_json_schema_version)
+          validator = validator(graphql_type_name, selected_schema_version)
 
           if (error_message = validator.validate_with_error_message(record))
             build_failed_result(event, "#{graphql_type_name} record", error_message)
@@ -130,7 +127,7 @@ def validate_record_returning_failure(event, selected_json_schema_version)
         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
+          # Here we use the `RecordPreparer::Identity` record preparer because we may not have a valid 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.
           operations = build_all_operations_for(event, RecordPreparer::Identity)

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

@@ -125,7 +125,7 @@ def calculate_latency_metrics(successful_operations, noop_results)
             "message_id" => event["message_id"],
             "event_type" => event.fetch("type"),
             "event_id" => EventID.from_event(event).to_s,
-            JSON_SCHEMA_VERSION_KEY => event.fetch(JSON_SCHEMA_VERSION_KEY),
+            SCHEMA_VERSION_KEY => event.fetch(SCHEMA_VERSION_KEY),
             "latencies_in_ms_from" => latencies_in_ms_from,
             "slo_results" => slo_results,
             "result" => result

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

@@ -11,7 +11,7 @@
 module ElasticGraph
   class Indexer
     class RecordPreparer
-      # Provides the ability to get a `RecordPreparer` for a specific JSON schema version.
+      # Provides the ability to get a `RecordPreparer` for a specific schema artifact version.
       class Factory
         def initialize(schema_artifacts)
           @schema_artifacts = schema_artifacts
@@ -21,23 +21,23 @@ def initialize(schema_artifacts)
             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|
+          @preparers_by_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
+        # Gets the `RecordPreparer` for the given schema artifact version.
+        def for_schema_version(schema_version)
+          @preparers_by_schema_version[schema_version] # : RecordPreparer
         end
 
-        # Gets the `RecordPreparer` for the latest JSON schema version. Intended primarily
+        # Gets the `RecordPreparer` for the latest schema artifact 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)
+        def for_latest_schema_version
+          for_schema_version(@schema_artifacts.latest_schema_version)
         end
 
         private