ElasticGraph::JSONIngestion

Provides JSON Schema ingestion support for ElasticGraph.

This gem provides the schema-definition extension that generates JSON Schema artifacts for indexing events and validates JSON-ingestion-specific schema options. Generated ElasticGraph projects install and enable it by default. Applications that wire schema-definition tasks manually enable it by adding ElasticGraph::JSONIngestion::SchemaDefinition::APIExtension to their schema-definition extension modules.

Schema Definition APIs

Use schema.json_schema_version to identify the current JSON schema artifact. Every change that affects the JSON schema should increment this version so publishers and indexers can safely evolve independently. During early prototyping, schema.enforce_json_schema_version false can disable version-change enforcement. For production applications, leave enforcement enabled so schema artifact changes require an intentional version bump.

diff --git a/config/schema.rb b/config/schema.rb
index 015c5fa..b8eeaef 100644
--- a/config/schema.rb
+++ b/config/schema.rb
@@ -2,3 +2,3 @@ ElasticGraph.define_schema do |schema|
   # ElasticGraph will tell you when you need to bump this.
-  schema.json_schema_version 1
+  schema.json_schema_version 2

# in config/schema/json_schema_enforcement.rb

ElasticGraph.define_schema do |schema|
  schema.enforce_json_schema_version false
end

Use schema.json_schema_strictness to configure whether indexing events may omit nullable fields or include extra fields. We recommend enabling at most one of these options, because enabling both can hide misspelled event fields.

# in config/schema/json_schema_strictness.rb

ElasticGraph.define_schema do |schema|
  schema.json_schema_strictness allow_omitted_fields: true, allow_extra_fields: false
end

Custom scalar types must declare how they are represented in JSON Schema:

# in config/schema/url.rb

ElasticGraph.define_schema do |schema|
  schema.scalar_type "URL" do |t|
    t.mapping type: "keyword"
    t.json_schema type: "string", format: "uri"
  end
end

Fields and object/interface types can add JSON Schema validations. Use field-level validations sparingly: they run while indexing events, so violations can send otherwise valid source-system data to the dead letter queue. They are best reserved for constraints that ElasticGraph needs in order to index correctly.

# in config/schema/card.rb

ElasticGraph.define_schema do |schema|
  schema.object_type "Card" do |t|
    t.field "id", "ID!"

    t.field "expYear", "Int" do |f|
      f.json_schema minimum: 2000, maximum: 2099
    end

    t.field "expMonth", "Int" do |f|
      f.json_schema minimum: 1, maximum: 12
    end

    t.index "cards"
  end
end

On fields, nullable: false disallows null in indexing events while keeping the GraphQL field nullable:

# in config/schema/widget.rb

ElasticGraph.define_schema do |schema|
  schema.object_type "Widget" do |t|
    t.field "id", "ID!"

    t.field "name", "String" do |f|
      f.json_schema nullable: false
    end

    t.index "widgets"
  end
end

Indexing Support

Beyond schema definition, this gem provides an adapter used by elasticgraph-indexer to ingest JSON events. The adapter validates each event envelope against the JSON schema identified by the event's json_schema_version, builds an ElasticGraph::Indexer::Event for each valid envelope, validates each record, and prepares records for indexing using that version's view of the schema.

JSON ingestion is enabled automatically for schemas defined with this gem's SchemaDefinition::APIExtension; no indexer configuration is needed.

This gem also provides the be_a_valid_elastic_graph_event RSpec matcher (via require "elastic_graph/json_ingestion/spec_support/event_matcher") for testing that publisher events conform to your schema.

Processing JSON Lines Payloads

ElasticGraph::JSONIngestion::Indexer wraps the format-neutral ElasticGraph::Indexer with JSON Lines decoding:

require "elastic_graph/json_ingestion/indexer"

indexer = ElasticGraph::JSONIngestion::Indexer.from_yaml_file("config/settings/local.yaml")
json_lines_payload = $stdin.read
indexer.process(json_lines_payload) unless json_lines_payload.empty?

Use #process_returning_failures when individual failures must be handled by the caller. It accepts decoded events, so a transport that must add metadata or combine several payloads into one bulk operation can call #decode first, adjust the decoded events, and then process them:

require "elastic_graph/json_ingestion/indexer"

indexer = ElasticGraph::JSONIngestion::Indexer.from_yaml_file("config/settings/local.yaml")

# One JSON Lines payload per blank-line-delimited chunk.
payloads = $stdin.read.split("\n\n")
decoded_events = payloads.flat_map { |payload| indexer.decode(payload) }

unless decoded_events.empty?
  failures = indexer.process_returning_failures(decoded_events)
  abort "#{failures.size} of #{decoded_events.size} event(s) failed." if failures.any?
end

The wrapper accepts an existing ElasticGraph::Indexer, so independent JSON and protobuf wrappers can share the same format-neutral indexer and its datastore clients:

require "elastic_graph/json_ingestion/indexer"

base_indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml")
json_indexer = ElasticGraph::JSONIngestion::Indexer.new(base_indexer)

Dependency Diagram

graph LR; classDef targetGemStyle fill:#FADBD8,stroke:#EC7063,color:#000,stroke-width:2px; classDef otherEgGemStyle fill:#A9DFBF,stroke:#2ECC71,color:#000; classDef externalGemStyle fill:#E0EFFF,stroke:#70A1D7,color:#2980B9; elasticgraph-json_ingestion["elasticgraph-json_ingestion"]; class elasticgraph-json_ingestion targetGemStyle; elasticgraph-indexer["elasticgraph-indexer"]; elasticgraph-json_ingestion --> elasticgraph-indexer; class elasticgraph-indexer otherEgGemStyle; elasticgraph-support["elasticgraph-support"]; elasticgraph-json_ingestion --> elasticgraph-support; class elasticgraph-support otherEgGemStyle; elasticgraph-indexer_lambda["elasticgraph-indexer_lambda"]; elasticgraph-indexer_lambda --> elasticgraph-json_ingestion; class elasticgraph-indexer_lambda otherEgGemStyle;