Add configurable indexing event decoder

config/settings/test.yaml.template

@@ -87,5 +87,8 @@ logger:
   device: log/test.log
 indexer:
   latency_slo_thresholds_by_timestamp_in_ms: {}
+  indexing_event_decoder:
+    name: ElasticGraph::JSONIngestion::IndexingEventDecoder
+    require_path: elastic_graph/json_ingestion/indexing_event_decoder
 schema_artifacts:
   directory: config/schema/artifacts

elasticgraph-indexer/README.md

@@ -48,3 +48,44 @@ indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml")
 events = [] # JSON events read from an async datastream
 indexer.processor.process(events)
 ```
+
+## Payload Decoding
+
+Ingesting encoded payloads from a transport (such as the SQS lambdas) requires an indexing event decoder extension,
+configured via the `indexer.indexing_event_decoder` setting. Decoders turn raw payload strings into ElasticGraph
+indexing event hashes before the normal validation and indexing pipeline runs. Ingestion format gems provide decoder
+implementations, or you can define your own:
+
+```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 this interface:
+
+```ruby
+# lib/my_company/elastic_graph/csv_indexing_event_decoder.rb
+module MyCompany
+  module ElasticGraph
+    class CSVIndexingEventDecoder
+      def initialize(config:, schema_artifacts:, logger:)
+        # `config` is a hash containing parameterized configuration values from the
+        # `indexing_event_decoder.config` setting (see above for an example).
+        #
+        # `schema_artifacts` provides access to the schema artifacts, in case decoding
+        # depends on the schema.
+        #
+        # `logger` is the ElasticGraph logger.
+      end
+
+      def decode(payload)
+        # Must return an array of ElasticGraph indexing event hashes.
+      end
+    end
+  end
+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"
 
@@ -86,6 +87,25 @@ def operation_factory
       end
     end
 
+    def indexing_event_decoder
+      @indexing_event_decoder ||= begin
+        extension = config.indexing_event_decoder
+
+        if extension.nil?
+          raise Errors::ConfigError, "`indexer.indexing_event_decoder` is not configured, but is required to " \
+            "decode indexing event payloads. Configure it with a decoder extension provided by an ingestion " \
+            "format gem (or one of your own) that understands your transport's payload format."
+        end
+
+        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,14 @@
 #
 # 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)
       json_schema at: "indexer",
         optional: false,
         description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.",
@@ -42,17 +44,67 @@ 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. " \
+              "Required when using a transport that delivers encoded payloads (such as the SQS lambdas). Ingestion " \
+              "format gems provide decoder implementations.",
+            type: ["object", "null"],
+            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: nil,
+            examples: [
+              {
+                "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)
+        return nil if config.nil?
+
+        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,34 @@
+# 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 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
+        def decode(payload)
+          # :nocov: -- must return an array to satisfy Steep type checking but never called
+          []
+          # :nocov:
+        end
+      end
+    end
+  end
+end

elasticgraph-indexer/sig/elastic_graph/indexer.rbs

@@ -28,6 +28,9 @@ module ElasticGraph
     @operation_factory: Operation::Factory?
     def operation_factory: () -> Operation::Factory
 
+    @indexing_event_decoder: indexingEventDecoder?
+    def indexing_event_decoder: () -> indexingEventDecoder
+
     @monotonic_clock: Support::MonotonicClock?
     def monotonic_clock: () -> Support::MonotonicClock
 

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

@@ -5,14 +5,17 @@ module ElasticGraph
 
       attr_reader latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer]
       attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]
+      attr_reader indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension?
 
       def initialize: (
         ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer],
-        ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]) -> void
+        ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
+        ?indexing_event_decoder: ::Hash[::String, untyped]?) -> void
 
       def with: (
         ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer],
-        ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]) -> Config
+        ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
+        ?indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension?) -> Config
 
       def self.members: () -> ::Array[::Symbol]
     end
@@ -22,8 +25,11 @@ module ElasticGraph
 
       def convert_values: (
         latency_slo_thresholds_by_timestamp_in_ms: untyped,
-        skip_derived_indexing_type_updates: untyped
+        skip_derived_indexing_type_updates: untyped,
+        indexing_event_decoder: untyped
       ) -> ::Hash[::Symbol, untyped]
+
+      def load_indexing_event_decoder: (::Hash[::String, untyped]?) -> SchemaArtifacts::RuntimeMetadata::Extension?
     end
   end
 end

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

@@ -0,0 +1,17 @@
+module ElasticGraph
+  class Indexer
+    type indexingEventDecoder = IndexingEventDecoder::Interface
+
+    module IndexingEventDecoder
+      class Interface
+        def initialize: (
+          config: ::Hash[::Symbol | ::String, untyped],
+          schema_artifacts: schemaArtifacts,
+          logger: ::Logger
+        ) -> void
+
+        def decode: (::String) -> ::Array[event]
+      end
+    end
+  end
+end

elasticgraph-indexer/spec/support/example_extensions/indexing_event_decoder.rb

@@ -0,0 +1,26 @@
+# 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
+
+class ExampleIndexingEventDecoder
+  attr_reader :config, :schema_artifacts, :logger
+
+  def initialize(config:, schema_artifacts:, logger:)
+    @config = config
+    @schema_artifacts = schema_artifacts
+    @logger = logger
+  end
+
+  def decode(payload)
+    payload.split(config.fetch("delimiter")).map { |value| {"value" => value} }
+  end
+end
+
+class InvalidIndexingEventDecoder
+  def initialize(config:, schema_artifacts:, logger:)
+  end
+end

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

@@ -31,6 +31,34 @@ class Indexer
 
         expect(config.skip_derived_indexing_type_updates).to eq("WidgetCurrency" => ["USD"].to_set)
       end
+
+      it "leaves the indexing event decoder unconfigured by default" do
+        expect(Config.new.indexing_event_decoder).to be nil
+      end
+
+      it "loads a configured indexing event decoder" do
+        config = Config.from_parsed_yaml("indexer" => {
+          "indexing_event_decoder" => {
+            "name" => "ExampleIndexingEventDecoder",
+            "require_path" => "support/example_extensions/indexing_event_decoder",
+            "config" => {"delimiter" => "|"}
+          }
+        })
+
+        expect(config.indexing_event_decoder.extension_class).to be(ExampleIndexingEventDecoder)
+        expect(config.indexing_event_decoder.config).to eq({"delimiter" => "|"})
+      end
+
+      it "verifies that a configured indexing event decoder implements the expected interface" do
+        expect {
+          Config.from_parsed_yaml("indexer" => {
+            "indexing_event_decoder" => {
+              "name" => "InvalidIndexingEventDecoder",
+              "require_path" => "support/example_extensions/indexing_event_decoder"
+            }
+          })
+        }.to raise_error Errors::InvalidExtensionError, a_string_including("InvalidIndexingEventDecoder", "Missing instance methods: `decode`")
+      end
     end
   end
 end

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

@@ -11,7 +11,12 @@
 module ElasticGraph
   RSpec.describe Indexer do
     it "returns non-nil values from each attribute" do
-      expect_to_return_non_nil_values_from_all_attributes(build_indexer)
+      indexer = build_indexer(indexing_event_decoder: {
+        "name" => "ExampleIndexingEventDecoder",
+        "require_path" => "support/example_extensions/indexing_event_decoder"
+      })
+
+      expect_to_return_non_nil_values_from_all_attributes(indexer)
     end
 
     describe ".from_parsed_yaml" do
@@ -29,5 +34,34 @@ module ElasticGraph
         expect(indexer).to be_a(Indexer)
       end
     end
+
+    describe "#indexing_event_decoder" do
+      it "builds the configured indexing event decoder" do
+        indexer = build_indexer(indexing_event_decoder: {
+          "name" => "ExampleIndexingEventDecoder",
+          "require_path" => "support/example_extensions/indexing_event_decoder",
+          "config" => {"delimiter" => "|"}
+        })
+
+        decoder = indexer.indexing_event_decoder
+
+        expect(decoder).to be_a(ExampleIndexingEventDecoder)
+        expect(decoder.config).to eq({"delimiter" => "|"})
+        expect(decoder.schema_artifacts).to be(indexer.schema_artifacts)
+        expect(decoder.logger).to be(indexer.logger)
+        expect(decoder.decode("one|two")).to eq([
+          {"value" => "one"},
+          {"value" => "two"}
+        ])
+      end
+
+      it "raises a clear error when no indexing event decoder is configured" do
+        indexer = build_indexer
+
+        expect {
+          indexer.indexing_event_decoder
+        }.to raise_error Errors::ConfigError, a_string_including("indexer.indexing_event_decoder")
+      end
+    end
   end
 end