diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index 218b6818d..f66fc63c3 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb @@ -63,6 +63,16 @@ def record_preparer_factory 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. + def ingestion_adapters + @ingestion_adapters ||= begin + require "elastic_graph/indexer/ingestion_adapter/json_events" + [IngestionAdapter::JSONEvents.new(schema_artifacts: schema_artifacts, logger: logger)] + end + end + def processor @processor ||= begin require "elastic_graph/indexer/processor" @@ -82,10 +92,9 @@ def operation_factory Operation::Factory.new( schema_artifacts: schema_artifacts, index_definitions_by_graphql_type: datastore_core.index_definitions_by_graphql_type, - record_preparer_factory: record_preparer_factory, + ingestion_adapters: ingestion_adapters, logger: datastore_core.logger, - skip_derived_indexing_type_updates: config.skip_derived_indexing_type_updates, - configure_record_validator: nil + skip_derived_indexing_type_updates: config.skip_derived_indexing_type_updates ) end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb index 133ff5d7d..5453553db 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb +++ b/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, logger + # @dynamic record_preparer_factory, processor, operation_factory, ingestion_adapters, logger # @dynamic self.from_parsed_yaml end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb new file mode 100644 index 000000000..7ac579154 --- /dev/null +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb @@ -0,0 +1,86 @@ +# 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 ingestion adapters. An ingestion adapter teaches the indexer how to handle + # events of a particular ingestion format: it validates each event and provides the + # version-appropriate machinery to prepare the event's record for indexing. + module IngestionAdapter + # Defines the ingestion adapter interface. Adapter classes are not required to subclass this, + # but must implement these methods. + class Interface + # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts + # @param logger [Logger] the ElasticGraph logger + def initialize(schema_artifacts:, logger:) + # must be defined, but nothing to do + end + + # Indicates whether this adapter recognizes the given event as one of its own. When multiple + # adapters are available, the indexer routes each event to the first adapter that returns + # `true`. (When exactly one adapter is available, it receives all events.) + # + # @param event [Hash] an ElasticGraph indexing event + # @return [Boolean] whether this adapter handles the event + def handles_event?(event) + # :nocov: -- must return a boolean to satisfy Steep type checking but never called + false + # :nocov: + end + + # Validates the given event and resolves the record preparer appropriate for the event's + # schema version. + # + # @param event [Hash] an ElasticGraph indexing event + # @return [ValidationResult] the result of validating the event + def validate_event(event) + # :nocov: -- must return a result to satisfy Steep type checking but never called + ValidationResult.valid(RecordPreparer::Identity) + # :nocov: + end + end + + # Describes a validation problem with an event. + # + # @!attribute [r] payload_description + # @return [String] brief description of the part of the event that was invalid + # @!attribute [r] message + # @return [String] detailed validation failure message + Failure = ::Data.define(:payload_description, :message) + + # Returned by {Interface#validate_event}. Either `failure` is non-nil (the event was invalid) + # or `record_preparer` is non-nil (the event was valid and its record can be prepared for + # indexing with the given preparer). + # + # @!attribute [r] record_preparer + # @return [Object, nil] preparer for the event's record, when the event is valid + # @!attribute [r] failure + # @return [Failure, nil] description of the validation problem, when the event is invalid + ValidationResult = ::Data.define(:record_preparer, :failure) do + # @implements ValidationResult + + # Builds a result for a valid event. + # + # @param record_preparer [Object] preparer for the event's record + # @return [ValidationResult] + def self.valid(record_preparer) + new(record_preparer: record_preparer, failure: nil) + end + + # Builds a result for an invalid event. + # + # @param payload_description [String] brief description of the part of the event that was invalid + # @param message [String] detailed validation failure message + # @return [ValidationResult] + def self.invalid(payload_description:, message:) + new(record_preparer: nil, failure: Failure.new(payload_description: payload_description, message: message)) + end + end + end + end +end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb new file mode 100644 index 000000000..fef2ba26b --- /dev/null +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb @@ -0,0 +1,138 @@ +# 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" +require "elastic_graph/indexer/ingestion_adapter" +require "elastic_graph/indexer/record_preparer" +require "elastic_graph/support/json_schema/validator_factory" + +module ElasticGraph + class Indexer + module IngestionAdapter + # Ingestion adapter for events in ElasticGraph's versioned JSON format: it validates events + # against the JSON schema identified by the event's `json_schema_version`, and prepares + # records using that version's view of the schema. + class JSONEvents + # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts + # @param logger [Logger] the ElasticGraph logger + # @param configure_record_validator [Proc, nil] optional callback to further configure the record validator + def initialize(schema_artifacts:, logger:, configure_record_validator: nil) + @schema_artifacts = schema_artifacts + @logger = logger + @configure_record_validator = configure_record_validator + @record_preparer_factory = RecordPreparer::Factory.new(schema_artifacts) + end + + # (see Interface#handles_event?) + def handles_event?(event) + event.key?(JSON_SCHEMA_VERSION_KEY) + end + + # (see Interface#validate_event) + def validate_event(event) + selected_json_schema_version = select_json_schema_version(event) { |failure| return failure } + + # 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}) + + if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope)) + return ValidationResult.invalid(payload_description: "event payload", message: error_message) + end + + record = event.fetch("record") + graphql_type_name = event.fetch("type") + + if (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record)) + return ValidationResult.invalid(payload_description: "#{graphql_type_name} record", message: error_message) + end + + ValidationResult.valid(@record_preparer_factory.for_json_schema_version(selected_json_schema_version)) + end + + private + + def select_json_schema_version(event) + available_json_schema_versions = @schema_artifacts.available_json_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 ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`") + elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1 + yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_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 + # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available + # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match, + # 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 + # 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 } + + if selected_json_schema_version != requested_json_schema_version + @logger.info({ + "message_type" => "ElasticGraphMissingJSONSchemaVersion", + "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 + }) + end + + if selected_json_schema_version.nil? + yield ValidationResult.invalid( + payload_description: JSON_SCHEMA_VERSION_KEY, + message: "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \ + Available json schema versions: #{available_json_schema_versions.sort.join(", ")}" + ) + end + + selected_json_schema_version + end + + def validator(type, selected_json_schema_version) + factory = validator_factories_by_version[selected_json_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| + factory = Support::JSONSchema::ValidatorFactory.new( + schema: @schema_artifacts.json_schemas_for(json_schema_version), + sanitize_pii: true + ) + + if (configure_record_validator = @configure_record_validator) + factory = configure_record_validator.call(factory) + end + + hash[json_schema_version] = factory + end + end + + # :nocov: -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class. + def raise(*args) + super("`raise` was called on `IngestionAdapter::JSONEvents`, but should not. Instead, use " \ + "`yield ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \ + "the valid events to still be processed.") + end + # :nocov: + end + end + end +end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb index c6b5365de..d68740bd2 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb @@ -6,12 +6,10 @@ # # frozen_string_literal: true -require "elastic_graph/constants" require "elastic_graph/indexer/event_id" require "elastic_graph/indexer/failed_event_error" require "elastic_graph/indexer/operation/update" require "elastic_graph/indexer/record_preparer" -require "elastic_graph/support/json_schema/validator_factory" require "elastic_graph/support/memoizable_data" module ElasticGraph @@ -20,94 +18,35 @@ module Operation class Factory < Support::MemoizableData.define( :schema_artifacts, :index_definitions_by_graphql_type, - :record_preparer_factory, + :ingestion_adapters, :logger, - :skip_derived_indexing_type_updates, - :configure_record_validator + :skip_derived_indexing_type_updates ) def build(event) event = prepare_event(event) - selected_json_schema_version = select_json_schema_version(event) { |failure| return failure } - - # 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}) - - if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope)) - return build_failed_result(event, "event payload", error_message) - end - - failed_result = validate_record_returning_failure(event, selected_json_schema_version) - failed_result || BuildResult.success(build_all_operations_for( - event, - record_preparer_factory.for_json_schema_version(selected_json_schema_version) - )) - end - - private - - def select_json_schema_version(event) - available_json_schema_versions = schema_artifacts.available_json_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 (adapter = ingestion_adapter_for(event)) + return build_failed_result(event, "event payload", "No available ingestion adapter recognized this event.") end - # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema - # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available - # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match, - # 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 - # 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 } - - if selected_json_schema_version != requested_json_schema_version - logger.info({ - "message_type" => "ElasticGraphMissingJSONSchemaVersion", - "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 - }) - end + validation_result = adapter.validate_event(event) - if selected_json_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(", ")}" - ) + if (failure = validation_result.failure) + return build_failed_result(event, failure.payload_description, failure.message) end - selected_json_schema_version + record_preparer = validation_result.record_preparer # : _RecordPreparer + BuildResult.success(build_all_operations_for(event, record_preparer)) end - def validator(type, selected_json_schema_version) - factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory - factory.validator_for(type) - end + private - def validator_factories_by_version - @validator_factories_by_version ||= ::Hash.new do |hash, json_schema_version| - factory = Support::JSONSchema::ValidatorFactory.new( - schema: schema_artifacts.json_schemas_for(json_schema_version), - sanitize_pii: true - ) - factory = configure_record_validator.call(factory) if configure_record_validator - hash[json_schema_version] = factory - end + # Routes the event to the first ingestion adapter that recognizes it. When exactly one + # adapter is available, it receives all events--including unrecognizable ones--so that + # its more specific validation failure messages are used. + def ingestion_adapter_for(event) + ingestion_adapters.find { |adapter| adapter.handles_event?(event) } || + (ingestion_adapters.first if ingestion_adapters.size == 1) end # This copies the `id` from event into the actual record @@ -117,22 +56,12 @@ 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) - record = event.fetch("record") - graphql_type_name = event.fetch("type") - validator = validator(graphql_type_name, selected_json_schema_version) - - if (error_message = validator.validate_with_error_message(record)) - build_failed_result(event, "#{graphql_type_name} record", error_message) - end - end - 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 - # 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. + # Here we use the `RecordPreparer::Identity` record preparer because the event failed validation, so + # no adapter-provided record preparer is available, and we won't wind up using the record preparer + # for real on these operations, anyway. operations = build_all_operations_for(event, RecordPreparer::Identity) BuildResult.failure(FailedEventError.new(event: event, operations: operations.to_set, main_message: message)) diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb index bdfc2791c..d1a705e59 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb @@ -6,14 +6,21 @@ # # frozen_string_literal: true +require "elastic_graph/indexer/ingestion_adapter/json_events" require "json" # Defines an RSpec matcher that can be used to validate ElasticGraph events. ::RSpec::Matchers.define :be_a_valid_elastic_graph_event do |for_indexer:| match do |event| + json_events_adapter = ElasticGraph::Indexer::IngestionAdapter::JSONEvents.new( + schema_artifacts: for_indexer.schema_artifacts, + logger: for_indexer.logger, + configure_record_validator: block_arg + ) + result = for_indexer .operation_factory - .with(configure_record_validator: block_arg) + .with(ingestion_adapters: [json_events_adapter]) .build(event) @validation_failure = result.failed_event_error diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs index 6f1d1bc40..c876f01d2 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs @@ -28,6 +28,9 @@ module ElasticGraph @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 diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs new file mode 100644 index 000000000..2335ea676 --- /dev/null +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter.rbs @@ -0,0 +1,42 @@ +module ElasticGraph + class Indexer + module IngestionAdapter + interface _Adapter + def handles_event?: (event) -> bool + def validate_event: (event) -> ValidationResult + end + + class Interface + include _Adapter + + def initialize: (schema_artifacts: schemaArtifacts, logger: ::Logger) -> void + end + + class FailureSupertype < ::Data + attr_reader payload_description: ::String + attr_reader message: ::String + + def initialize: (payload_description: ::String, message: ::String) -> void + def self.new: (payload_description: ::String, message: ::String) -> instance + def self.members: () -> ::Array[::Symbol] + end + + class Failure < FailureSupertype + end + + class ValidationResultSupertype < ::Data + attr_reader record_preparer: _RecordPreparer? + attr_reader failure: Failure? + + def initialize: (record_preparer: _RecordPreparer?, failure: Failure?) -> void + def self.new: (record_preparer: _RecordPreparer?, failure: Failure?) -> instance + def self.members: () -> ::Array[::Symbol] + end + + class ValidationResult < ValidationResultSupertype + def self.valid: (_RecordPreparer) -> ValidationResult + def self.invalid: (payload_description: ::String, message: ::String) -> ValidationResult + end + end + end +end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs new file mode 100644 index 000000000..a10b69923 --- /dev/null +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs @@ -0,0 +1,31 @@ +module ElasticGraph + class Indexer + module IngestionAdapter + class JSONEvents + def initialize: ( + schema_artifacts: schemaArtifacts, + logger: ::Logger, + ?configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)? + ) -> void + + @schema_artifacts: schemaArtifacts + @logger: ::Logger + @configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)? + @record_preparer_factory: RecordPreparer::Factory + + def handles_event?: (event) -> bool + def validate_event: (event) -> ValidationResult + + private + + def select_json_schema_version: (event) { (ValidationResult) -> bot } -> (::Integer | bot) + def validator: (::String, ::Integer) -> Support::JSONSchema::Validator + + @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]? + def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory] + + def raise: (*untyped) -> bot + end + end + end +end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs index 82e968053..2dfc73280 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs @@ -1,34 +1,29 @@ module ElasticGraph class Indexer type event = ::Hash[::String, untyped] - type validator = Support::JSONSchema::Validator - type validatorFactory = Support::JSONSchema::ValidatorFactory module Operation class FactorySupertype attr_reader schema_artifacts: schemaArtifacts attr_reader index_definitions_by_graphql_type: ::Hash[::String, ::Array[DatastoreCore::_IndexDefinition]] - attr_reader record_preparer_factory: RecordPreparer::Factory + attr_reader ingestion_adapters: ::Array[IngestionAdapter::_Adapter] attr_reader logger: ::Logger attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] - attr_reader configure_record_validator: (^(validatorFactory) -> validatorFactory)? def initialize: ( schema_artifacts: schemaArtifacts, index_definitions_by_graphql_type: ::Hash[::String, ::Array[DatastoreCore::_IndexDefinition]], - record_preparer_factory: RecordPreparer::Factory, + ingestion_adapters: ::Array[IngestionAdapter::_Adapter], logger: ::Logger, - skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], - configure_record_validator: (^(validatorFactory) -> validatorFactory)? + skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] ) -> void def with: ( ?schema_artifacts: schemaArtifacts, ?index_definitions_by_graphql_type: ::Hash[::String, ::Array[DatastoreCore::_IndexDefinition]], - ?record_preparer_factory: RecordPreparer::Factory, + ?ingestion_adapters: ::Array[IngestionAdapter::_Adapter], ?logger: ::Logger, - ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], - ?configure_record_validator: (^(validatorFactory) -> validatorFactory)? + ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] ) -> instance end @@ -37,14 +32,8 @@ module ElasticGraph private - def validator: (::String, ::Integer) -> Support::JSONSchema::Validator - - @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]? - def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory] - - def select_json_schema_version: (event) { (BuildResult) -> bot } -> (::Integer | bot) + def ingestion_adapter_for: (event) -> IngestionAdapter::_Adapter? def prepare_event: (event) -> event - def validate_record_returning_failure: (event, ::Integer) -> BuildResult? def build_failed_result: (event, ::String, ::String) -> BuildResult def build_all_operations_for: (event, _RecordPreparer) -> ::Array[_Operation] def index_definitions_for: (::String) -> ::Array[DatastoreCore::_IndexDefinition] diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb index 1fed90d64..80608dbce 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb @@ -6,8 +6,9 @@ # # frozen_string_literal: true -require "elastic_graph/indexer" require "elastic_graph/constants" +require "elastic_graph/indexer" +require "elastic_graph/indexer/ingestion_adapter/json_events" require "elastic_graph/indexer/operation/factory" require "elastic_graph/spec_support/builds_indexer_operation" require "json" @@ -391,6 +392,58 @@ module Operation expect(operations.map(&:doc_id)).to contain_exactly("c1", "c2", "c3") end + context "when multiple ingestion adapters are available" do + it "routes each event to the first adapter that recognizes it" do + event = build_upsert_event(:component, id: "1", __version: 1) + + non_matching_adapter = instance_double(IngestionAdapter::Interface, handles_event?: false) + matching_adapter = instance_double( + IngestionAdapter::Interface, + handles_event?: true, + validate_event: IngestionAdapter::ValidationResult.valid(RecordPreparer::Identity) + ) + + factory = indexer.operation_factory.with(ingestion_adapters: [non_matching_adapter, matching_adapter]) + result = factory.build(event) + + expect(result.failed_event_error).to be nil + expect(result.operations).not_to be_empty + expect(matching_adapter).to have_received(:validate_event).with(a_hash_including("type" => "Component")) + end + + it "fails the event when no adapter recognizes it" do + event = build_upsert_event(:component, id: "1", __version: 1) + + adapters = [ + instance_double(IngestionAdapter::Interface, handles_event?: false), + instance_double(IngestionAdapter::Interface, handles_event?: false) + ] + + factory = indexer.operation_factory.with(ingestion_adapters: adapters) + + expect_failed_event_error(event, "No available ingestion adapter recognized this event.", factory: factory) + end + end + + context "when a single ingestion adapter is available" do + it "routes all events to it, even ones it does not recognize, so that its more specific failure messages are used" do + event = build_upsert_event(:component, id: "1", __version: 1) + + adapter = instance_double( + IngestionAdapter::Interface, + handles_event?: false, + validate_event: IngestionAdapter::ValidationResult.invalid( + payload_description: "event payload", + message: "not recognizable by this adapter" + ) + ) + + factory = indexer.operation_factory.with(ingestion_adapters: [adapter]) + + expect_failed_event_error(event, "not recognizable by this adapter", factory: factory) + end + end + def expect_failed_event_error(event, *error_message_snippets, factory: indexer.operation_factory, expect_no_ops: false) result = factory.build(event) @@ -432,10 +485,19 @@ def event_id_from(event) end def build_expecting_success(event, **options, &configure_record_validator) - result = indexer - .operation_factory - .with(configure_record_validator: configure_record_validator) - .build(event, **options) + factory = indexer.operation_factory + + if configure_record_validator + json_events_adapter = IngestionAdapter::JSONEvents.new( + schema_artifacts: indexer.schema_artifacts, + logger: indexer.logger, + configure_record_validator: configure_record_validator + ) + + factory = factory.with(ingestion_adapters: [json_events_adapter]) + end + + result = factory.build(event, **options) raise result.failed_event_error if result.failed_event_error result.operations