Conversation
jwils
force-pushed
the
joshuaw/proto-integration
branch
4 times, most recently
from
September 15, 2026 14:20
49058c8 to
0ad7f71
Compare
- Add MalformedEventError for a payload whose envelope an ingestion adapter cannot make sense of, since it never reaches the point of becoming an event. The payload type is a type parameter, so each ingestion format keeps its own payload type. - Add JSONIngestion::EnvelopeValidator, which validates each decoded event's envelope and returns the valid events alongside a MalformedEventError per malformed one. It also owns the versioned JSON schema validators, which IngestionAdapter uses for record validation. Envelope validation is not part of the ingestion adapter interface and is only called from within the JSON ingestion gem, so it lives in its own class rather than on the adapter. - IngestionAdapter#validate_event now only validates the record, and exposes its envelope validator so callers can reach it. - JSONIngestion::Indexer#process_returning_failures takes decoded events and runs them through the envelope validator before the operation factory. This is the shape Indexer::Processor#process_returning_failures accepted in v1.2.0, so each layer takes events in the form that is natural for it: decoded JSON for the JSON indexer, built events for the processor. `process` decodes a JSON Lines payload and routes through it. - IndexerLambda::SqsProcessor: rename events_from to decoded_events_from and decoded_events_from to decode; merge SQS metadata into decoded events before envelope validation; call indexer.process_returning_failures. - Operation::Factory#build_all_operations_for fetches `type` rather than guarding for its absence, since the envelope is always valid by the time an event reaches the factory. - The JSON ingestion README documents the decoded-events entry point, rather than pointing transports at indexer.processor, which cannot accept decoded events. Generated with Claude Code
- Introduce Indexer::Event, a Data value object with a validated
envelope (op, type, id, version, record, schema_version,
ingestion_format, message_id, latency_timestamps).
- Swap every event["x"] / event.fetch("x") for typed accessors across
FailedEventError, IndexingFailuresError, Operation::Factory and
Update, DatastoreIndexingRouter, Processor, the JSON ingestion
adapter and EnvelopeValidator, WarehouseDumper,
TestSupport::Converters, and the affected specs and RBS files.
- Event#event_id returns the EventID for a validated event, so callers
ask the event for its id rather than building one from its fields.
EventID.from_decoded_hash covers the one case with no Event to ask:
a payload that failed envelope validation, where any envelope field
can be absent. Both keep the "type:id@vversion" format in EventID.
- FailedEventError delegates id/op/type/version/record/message_id to
its event via Forwardable instead of manual wrapper methods, and
includes IndexingFailuresError::_IndexingFailure so Steep enforces
the interface it already satisfies.
- EnvelopeValidator#events_from now returns Events via
Event.from_validated_hash instead of hashes.
- TestSupport::Converters keeps upsert_event_hash_for alongside
upsert_event_for, since the JSON hash and the Event are now distinct:
specs that exercise envelope handling need the hash.
- Add a JSONIngestion::Indexer spec proving envelope validation still
applies to a type configured to skip record validation, since that
guarantee now depends on events_from and validate_event being
separate steps rather than an explicit check in the operation
factory.
Generated with Claude Code
Add a native ingestion adapter and descriptor-based decoding for raw messages with transport metadata and generated event batches. Preserve protobuf presence, validate ElasticGraph constraints, and support nested lists and source-only updates without JSON schema versions. Run shared indexer integration tests through both wire formats and cover SQS, warehouse output, and live datastore queries.
jwils
force-pushed
the
joshuaw/proto-integration
branch
from
September 21, 2026 02:07
0ad7f71 to
474ca66
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
ElasticGraph can generate protobuf schemas, but publishers must convert protobuf messages to JSON before indexing them.
What
Add protobuf ingestion through the shared indexing pipeline. Publishers can send generated event batches or raw domain messages with transport metadata.
How
Generate a protobuf event envelope and private indexing metadata. The protobuf indexer validates each decoded envelope, builds a typed event, loads a descriptor set, validates the record, and prepares it with format-neutral field metadata. Shared integration tests exercise JSON and protobuf ingestion, including warehouse output.
Risk
Moderate. The protobuf extension is unreleased. Shared changes let adapters return typed events, pass transport metadata to decoders, and process events without a JSON schema version. Existing JSON warehouse partitions remain unchanged.
Testing
No manual testing.
Bigger picture
This branch is rebased on #1398. Keep the PR base on
main, and merge #1398 first.