Route events through a typed Event value

elasticgraph-indexer/README.md

@@ -48,6 +48,8 @@ require "elastic_graph/indexer"
 
 indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml")
 
-events = [] # JSON events read from an async datastream
+# `ElasticGraph::Indexer::Event` instances, built by an ingestion adapter such as the one
+# in `elasticgraph-json_ingestion` after it validates the event envelope.
+events = []
 indexer.processor.process(events)
 ```

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

@@ -68,7 +68,7 @@ def bulk(operations, refresh: false)
         unless unsupported_ops.empty?
           raise IndexingFailuresError,
             "The index definitions for #{unsupported_ops.size} operations " \
-            "(#{unsupported_ops.map { |o| Indexer::EventID.from_event(o.event) }.join(", ")}) " \
+            "(#{unsupported_ops.map { |o| o.event.event_id }.join(", ")}) " \
             "were configured to be inaccessible. Check the configuration, or avoid sending " \
             "events of this type to this ElasticGraph indexer."
         end
@@ -269,7 +269,7 @@ def source_event_versions_in_index(operations)
 
       def opaque_id_parts_for_source_event_versions(operations)
         type_counts = operations
-          .group_by { |op| op.event.fetch("type") }
+          .group_by { |op| op.event.type }
           .sort_by(&:first)
           .map { |type_name, ops| "#{type_name}:#{ops.size}" }
 

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

@@ -0,0 +1,74 @@
+# 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"
+
+module ElasticGraph
+  class Indexer
+    # An indexing event with a validated envelope. Ingestion adapters build events after they
+    # validate the envelope in their own format, so every `Event` instance has typed envelope
+    # fields. `record` keeps the adapter's native record type (a `Hash` for JSON events).
+    #
+    # @!attribute [r] op
+    #   @return [String] the operation to apply (`upsert`)
+    # @!attribute [r] type
+    #   @return [String] the GraphQL type of the record
+    # @!attribute [r] id
+    #   @return [String] the unique identifier of the record
+    # @!attribute [r] version
+    #   @return [Integer] version used to order events for the same `type` and `id`
+    # @!attribute [r] record
+    #   @return [Object] the record payload, in the ingestion adapter's native type
+    # @!attribute [r] schema_version
+    #   @return [Integer] the version of the ingestion schema the publisher used
+    # @!attribute [r] ingestion_format
+    #   @return [String] the format tag that selects the ingestion adapter
+    # @!attribute [r] message_id
+    #   @return [String, nil] the id of the transport message that carried the event
+    # @!attribute [r] latency_timestamps
+    #   @return [Hash<String, String>] ISO8601 timestamps from which indexing latency is measured
+    Event = ::Data.define(:op, :type, :id, :version, :record, :schema_version, :ingestion_format, :message_id, :latency_timestamps) do
+      # @implements Event[R]
+
+      def initialize(op:, type:, id:, version:, record:, schema_version:, ingestion_format:, message_id: nil, latency_timestamps: {})
+        super
+      end
+
+      # @return [EventID] identifies this event by its `type`, `id`, and `version`
+      def event_id
+        EventID.new(type: type, id: id, version: version)
+      end
+
+      # Builds an event from a JSON event hash whose envelope has already been validated.
+      #
+      # @param hash [Hash<String, Object>] a validated JSON event
+      # @return [Event]
+      def self.from_validated_hash(hash)
+        latency_timestamps = hash["latency_timestamps"] || {} # : ::Hash[::String, ::String]
+
+        new(
+          op: hash.fetch("op"),
+          type: hash.fetch("type"),
+          id: hash.fetch("id"),
+          version: hash.fetch("version"),
+          record: hash.fetch("record"),
+          schema_version: hash.fetch(JSON_SCHEMA_VERSION_KEY),
+          ingestion_format: hash.fetch(INGESTION_FORMAT_KEY, "json"),
+          message_id: hash["message_id"],
+          latency_timestamps: latency_timestamps
+        )
+      end
+    end
+
+    # Steep weirdly expects them here...
+    # @dynamic initialize, config, datastore_core, schema_artifacts, datastore_router, monotonic_clock
+    # @dynamic processor, operation_factory, ingestion_adapters_by_format, logger
+    # @dynamic self.from_parsed_yaml
+  end
+end

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

@@ -15,8 +15,14 @@ class Indexer
     # easy to put these ids in a comma-separated list.
     EventID = ::Data.define(:type, :id, :version) do
       # @implements EventID
-      def self.from_event(event)
-        new(type: event["type"], id: event["id"], version: event["version"])
+
+      # Builds an id from a decoded payload whose envelope is not yet known to be valid, so no
+      # {Event} exists for it. Use {Event#event_id} for a validated event.
+      #
+      # @param hash [Hash<String, Object>] a decoded indexing payload
+      # @return [EventID]
+      def self.from_decoded_hash(hash)
+        new(type: hash["type"], id: hash["id"], version: hash["version"])
       end
 
       def to_s

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

@@ -8,14 +8,17 @@
 
 require "elastic_graph/errors"
 require "elastic_graph/indexer/event_id"
+require "forwardable"
 
 module ElasticGraph
   class Indexer
     # Indicates an event that we attempted to process which failed for some reason. It may have
     # failed due to a validation issue before we even attempted to write it to the datastore, or it
     # could have failed in the datastore itself.
     class FailedEventError < Errors::Error
-      # @dynamic main_message, event, operations, message_id, message
+      extend ::Forwardable
+
+      # @dynamic main_message, event, operations, id, op, type, version, record, message_id, message
 
       # The "main" part of the error message (without the `full_id` portion).
       attr_reader :main_message
@@ -51,37 +54,15 @@ def versioned_operations
       end
 
       def full_id
-        event_id = EventID.from_event(event).to_s
-        if (message_id = self.message_id)
+        event_id = event.event_id.to_s
+        if (message_id = event.message_id)
           "#{event_id} (message_id: #{message_id})"
         else
           event_id
         end
       end
 
-      def id
-        event["id"]
-      end
-
-      def op
-        event["op"]
-      end
-
-      def type
-        event["type"]
-      end
-
-      def version
-        event["version"]
-      end
-
-      def record
-        event["record"]
-      end
-
-      def message_id
-        event["message_id"]
-      end
+      def_delegators :event, :id, :op, :type, :version, :record, :message_id
     end
   end
 end

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

@@ -15,13 +15,13 @@ module IngestionAdapter
       # Defines the ingestion adapter interface. Adapter classes are not required to subclass this,
       # but must implement these methods.
       class Interface
-        # Validates the given event and resolves the record preparer appropriate for the event's
-        # ingestion schema version. The indexer selects this adapter from the event's format tag.
-        # The indexer builds operations from the event in the returned result, so an adapter can
-        # return a prepared copy of the event.
+        # Validates the record of the given event and resolves the record preparer appropriate for
+        # the event's ingestion schema version. The indexer selects this adapter from the event's
+        # format tag. The indexer builds operations from the event in the returned result, so an
+        # adapter can return a prepared copy of the event.
         #
-        # @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
+        # @param event [Event] an ElasticGraph indexing event with a validated envelope
+        # @param skip_record_validation [Boolean] whether to skip record validation
         # @return [ValidationResult] the result of validating the event
         def validate_event(event, skip_record_validation: false)
           # simplecov:disable -- must return a result to satisfy Steep type checking but never called
@@ -42,7 +42,7 @@ def validate_event(event, skip_record_validation: false)
       # and a non-nil `record_preparer` indicates a valid event.
       #
       # @!attribute [r] event
-      #   @return [Hash<String, Object>, nil] the event to build operations from, when the event is valid
+      #   @return [Event, nil] the event to build operations from, when the event is valid
       # @!attribute [r] record_preparer
       #   @return [Object, nil] preparer for the event's record, when the event is valid
       # @!attribute [r] failure
@@ -52,7 +52,7 @@ def validate_event(event, skip_record_validation: false)
 
         # Builds a result for a valid event.
         #
-        # @param event [Hash<String, Object>] the event to build operations from; an adapter may return a prepared copy
+        # @param event [Event] the event to build operations from; an adapter may return a prepared copy
         # @param record_preparer [Object] preparer for the event's record
         # @return [ValidationResult]
         def self.valid(event, record_preparer)

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

@@ -25,7 +25,7 @@ class Factory < Support::MemoizableData.define(
         :skip_record_validation_percents_by_type
       )
         def build(event)
-          format = event.fetch(INGESTION_FORMAT_KEY, "json")
+          format = event.ingestion_format
           adapter = ingestion_adapters_by_format[format]
 
           unless adapter
@@ -43,7 +43,7 @@ def build(event)
         private
 
         def build_for_adapter(event, adapter)
-          skip_record_validation = skip_validation?(event["type"], event)
+          skip_record_validation = skip_validation?(event)
           validation_result = adapter.validate_event(event, skip_record_validation: skip_record_validation)
 
           if (failure = validation_result.failure)
@@ -75,7 +75,7 @@ def build_success_result(event, record_preparer, type_with_skipped_validation:)
         # 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, record_preparer, adapter)
-          build_success_result(event, record_preparer, type_with_skipped_validation: event.fetch("type"))
+          build_success_result(event, record_preparer, type_with_skipped_validation: event.type)
         rescue => exception
           failure = adapter.validate_event(event).failure
           # `raise` is overridden below to stop this class from *originating* an error instead of returning a
@@ -89,18 +89,18 @@ def build_success_result_isolating_malformed_records(event, record_preparer, ada
         # configured percent with a single multiply instead of dividing on every event.
         CRC32_SPACE_PER_PERCENT = (1 << 32) / 100.0
 
-        # Decides whether to skip per-record validation for `event` of `type`. The decision is
+        # Decides whether to skip per-record validation for `event`. The decision is
         # deterministic per event id: a stable `Zlib.crc32` of `EventID#to_s` maps each event to a
         # point in the CRC32 space, and we skip validation for the configured percentage of that
         # space. Same event id => same decision across pods and retries, so retries never flip a
         # record between validated and skipped. `String#hash` is unsuitable here, as `RUBY_HASH_SEED`
         # is per-process. The `<= 0` and `>= 100` guards keep the endpoints exact, so no float
         # boundary error can make a `0` percent skip a record or a `100` percent validate one.
-        def skip_validation?(type, event)
-          percent = skip_record_validation_percents_by_type[type]
+        def skip_validation?(event)
+          percent = skip_record_validation_percents_by_type[event.type]
           return false if percent.nil? || percent <= 0
           return true if percent >= 100
-          ::Zlib.crc32(EventID.from_event(event).to_s) < percent * CRC32_SPACE_PER_PERCENT
+          ::Zlib.crc32(event.event_id.to_s) < percent * CRC32_SPACE_PER_PERCENT
         end
 
         def build_failed_result(event, validation_target, validation_message)
@@ -120,8 +120,8 @@ def build_failed_result(event, validation_target, validation_message)
           rescue => exception
             logger.warn({
               "message_type" => "FailedEventOperationBuildingFailure",
-              "message_id" => event["message_id"],
-              "event_id" => EventID.from_event(event).to_s,
+              "message_id" => event.message_id,
+              "event_id" => event.event_id.to_s,
               "error_class" => exception.class.name,
               "error_message" => exception.message
             })
@@ -135,8 +135,7 @@ def build_all_operations_for(event, record_preparer)
           # If `type` is not a known type (as indicated by `runtime_metadata` being nil)
           # then we can't build a derived indexing type update operation. That case will only happen when we build
           # operations for an `FailedEventError` rather than to execute.
-          type = event.fetch("type")
-          return [] unless (runtime_metadata = schema_artifacts.runtime_metadata.object_types_by_name[type])
+          return [] unless (runtime_metadata = schema_artifacts.runtime_metadata.object_types_by_name[event.type])
 
           runtime_metadata.update_targets.flat_map do |update_target|
             ids_to_skip = skip_derived_indexing_type_updates.fetch(update_target.type, ::Set.new)
@@ -155,10 +154,10 @@ def build_all_operations_for(event, record_preparer)
                   if skipped
                     logger.info({
                       "message_type" => "SkippingUpdate",
-                      "message_id" => event["message_id"],
+                      "message_id" => event.message_id,
                       "update_target" => update_target.type,
                       "id" => op.doc_id,
-                      "event_id" => EventID.from_event(event).to_s
+                      "event_id" => event.event_id.to_s
                     })
                   end
                 end

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

@@ -48,7 +48,7 @@ def event
         end
 
         def event_id
-          EventID.from_event(event)
+          event.event_id
         end
 
         def summary

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

@@ -28,8 +28,8 @@ def self.operations_for(
           destination_index_mapping:
         )
           prepared_record = record_preparer.prepare_for_index(
-            event["type"],
-            event["record"] || {"id" => event["id"]},
+            event.type,
+            event.record,
             destination_index_mapping.fetch("properties")
           )
 
@@ -78,15 +78,15 @@ def type
         end
 
         def description
-          if update_target.type == event.fetch("type")
+          if update_target.type == event.type
             "#{update_target.type} update"
           else
-            "#{update_target.type} update (from #{event.fetch("type")})"
+            "#{update_target.type} update (from #{event.type})"
           end
         end
 
         def inspect
-          "#<#{self.class.name} event=#{EventID.from_event(event)} target=#{update_target.type}>"
+          "#<#{self.class.name} event=#{event.event_id} target=#{update_target.type}>"
         end
         alias_method :to_s, :inspect
 
@@ -141,7 +141,7 @@ def message_from_thrown_painless_exception(update)
         def script_params
           initial_params = update_target.params_for(
             doc_id: doc_id,
-            event: event,
+            event: event.to_h.transform_keys(&:to_s),
             prepared_record: prepared_record
           )
 

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

@@ -109,7 +109,7 @@ def categorize_failures(failures, events)
         end
 
         if superseded_failures.any?
-          superseded_ids = superseded_failures.map { |f| EventID.from_event(f.event).to_s }
+          superseded_ids = superseded_failures.map { |f| f.event.event_id.to_s }
           @logger.warn(
             "Ignoring #{superseded_ids.size} malformed event(s) because they have been superseded " \
             "by corrected events targeting the same id: #{superseded_ids.join(", ")}."
@@ -129,7 +129,7 @@ def calculate_latency_metrics(successful_operations, noop_results)
           latencies_in_ms_from = {} # : Hash[String, Integer]
           slo_results = {} # : Hash[String, String]
 
-          latency_timestamps = event.fetch("latency_timestamps", _ = {})
+          latency_timestamps = event.latency_timestamps
           latency_timestamps.each do |ts_name, ts_value|
             metric_value = ((current_time - Time.iso8601(ts_value)) * 1000).round
 
@@ -144,10 +144,10 @@ def calculate_latency_metrics(successful_operations, noop_results)
 
           @logger.info({
             "message_type" => "ElasticGraphIndexingLatencies",
-            "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),
+            "message_id" => event.message_id,
+            "event_type" => event.type,
+            "event_id" => event.event_id.to_s,
+            "schema_version" => event.schema_version,
             "latencies_in_ms_from" => latencies_in_ms_from,
             "slo_results" => slo_results,
             "result" => result