block · jwils · Jan 20, 2026 · Jan 17, 2026 · Jan 19, 2026 · Jan 19, 2026
@@ -42,6 +42,7 @@ bundle exec rake opensearch:test:boot
 
 ### Documentation
 - API documentation uses YARD
+- 100% documentation coverage is required for all public methods and classes.
 - Website source: `config/site/`
 - Example queries: `config/site/examples/*/queries/`
 - When writing links in documentation, use permalinks (links to a specific commit/version)
@@ -152,6 +153,9 @@ Custom gems can be added via `Gemfile-custom` (see `Gemfile-custom.example`), th
 - Prefer `::Data.define` over `::Struct.new` for immutable data classes. Use `Struct` only when mutability is required.
 - Don't rely on exceptions for control flow (exceptions are slow). Handle edge cases explicitly instead (e.g., check for `nil` before calling a method that would raise `ArgumentError`).
 - For constants accessed from multiple EG gems, define them in `elasticgraph-support/lib/elastic_graph/constants.rb`.
+- Always put a blank space after `#` in comments. This applies to all comments, including RBS type annotation comments (e.g., `# : String` not `#: String`).
+- Avoid defensive code for impossible cases. Don't add checks, tests, or handling for scenarios that should never occur due to the design of the system. If something can only happen due to a bug in the implementation, it's better to fail fast than to silently handle it.
+- Drop unnecessary namespace prefixes. When code is already inside `module ElasticGraph`, use `Indexer` instead of `::ElasticGraph::Indexer`, and `Errors::ConfigError` instead of `::ElasticGraph::Errors::ConfigError`. Use fully qualified names only when necessary to avoid ambiguity.
 
 ### RBS Type Signatures
 
@@ -160,6 +164,9 @@ Custom gems can be added via `Gemfile-custom` (see `Gemfile-custom.example`), th
 ### Testing
 
 - Avoid duplicate tests. If two tests will always pass/fail together, keep only one.
+- Use `expect_to_return_non_nil_values_from_all_attributes` to test wrapper classes (like `WarehouseLambda`, `GraphQL`, `Indexer`, etc.). This automatically exercises every zero-argument method and verifies all dependencies are built successfully.
+- Use the `:capture_logs` RSpec tag instead of logger test doubles for verifying log output. Access logs with `logged_jsons_of_type(message_type)`.
+- Use `build_*` helper methods from `spec/support/builds_*.rb` to construct test objects. These helpers provide sensible defaults while allowing selective overrides for testing specific scenarios.
 
 ## Important Patterns
 

@@ -80,7 +80,8 @@ target :elasticgraph_gems do
     "tmpdir",
     "tempfile",
     "time",
-    "uri"
+    "uri",
+    "zlib"
 
   configure_code_diagnostics(::Steep::Diagnostic::Ruby.all_error) do |config|
     # Setting these to :hint for now, as some branches are unreachable by steep

@@ -11,13 +11,13 @@ module ElasticGraph
     def initialize: (
       config: Config,
       datastore_core: DatastoreCore,
-      ?datastore_router: DatastoreIndexingRouter?,
+      ?datastore_router: Indexer::_DatastoreRouter?,
       ?monotonic_clock: Support::MonotonicClock?,
       ?clock: singleton(::Time)?
     ) -> void
 
-    @datastore_router: DatastoreIndexingRouter?
-    def datastore_router: () -> DatastoreIndexingRouter
+    @datastore_router: Indexer::_DatastoreRouter?
+    def datastore_router: () -> Indexer::_DatastoreRouter
 
     @record_preparer_factory: RecordPreparer::Factory?
     def record_preparer_factory: () -> RecordPreparer::Factory

@@ -1,14 +1,20 @@
 module ElasticGraph
   class Indexer
+    # Interface for routing indexing operations to a datastore (or alternative destination like S3).
+    # Implemented by DatastoreIndexingRouter and WarehouseDumper.
+    interface _DatastoreRouter
+      def bulk: (::Array[_Operation], ?refresh: bool) -> DatastoreIndexingRouter::BulkResult
+      def source_event_versions_in_index: (::Array[_Operation]) -> ::Hash[_Operation, ::Hash[::String, ::Array[::Integer]]]
+    end
+
     class DatastoreIndexingRouter
+      include _DatastoreRouter
+
       def initialize: (
         datastore_clients_by_name: ::Hash[::String, DatastoreCore::_Client],
         logger: ::Logger
       ) -> void
 
-      def bulk: (::Array[_Operation], ?refresh: bool) -> BulkResult
-      def source_event_versions_in_index: (::Array[_Operation]) -> ::Hash[_Operation, ::Hash[::String, ::Array[::Integer]]]
-
       private
 
       @datastore_clients_by_name: ::Hash[::String, DatastoreCore::_Client]

@@ -2,7 +2,7 @@ module ElasticGraph
   class Indexer
     class Processor
       def initialize: (
-        datastore_router: DatastoreIndexingRouter,
+        datastore_router: _DatastoreRouter,
         operation_factory: Operation::Factory,
         logger: ::Logger,
         indexing_latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer],
@@ -14,7 +14,7 @@ module ElasticGraph
 
       private
 
-      @datastore_router: DatastoreIndexingRouter
+      @datastore_router: _DatastoreRouter
       @operation_factory: Operation::Factory
       @logger: ::Logger
       @indexing_latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer]

@@ -11,12 +11,12 @@
 require "elastic_graph/elasticsearch/client"
 require "elastic_graph/indexer/datastore_indexing_router"
 require "elastic_graph/indexer/operation/factory"
-require "support/primary_indexing_operation_support"
+require "elastic_graph/spec_support/builds_indexer_operation"
 
 module ElasticGraph
   class Indexer
     RSpec.describe DatastoreIndexingRouter, :capture_logs do
-      include PrimaryIndexingOperationSupport
+      include SpecSupport::BuildsIndexerOperation
 
       let(:main_datastore_client) { instance_spy(Elasticsearch::Client, cluster_name: "main") }
       let(:other_datastore_client) { instance_spy(Elasticsearch::Client, cluster_name: "other") }

@@ -9,15 +9,15 @@
 require "elastic_graph/indexer"
 require "elastic_graph/constants"
 require "elastic_graph/indexer/operation/factory"
-require "support/primary_indexing_operation_support"
+require "elastic_graph/spec_support/builds_indexer_operation"
 require "json"
 
 module ElasticGraph
   class Indexer
     module Operation
       RSpec.describe Factory, :capture_logs do
         describe "#build", :factories do
-          include PrimaryIndexingOperationSupport
+          include SpecSupport::BuildsIndexerOperation
 
           let(:indexer) { build_indexer }
           let(:component_index_definition) { index_def_named("components") }

@@ -11,14 +11,14 @@
 require "elastic_graph/indexer/processor"
 require "elastic_graph/indexer/datastore_indexing_router"
 require "elastic_graph/support/hash_util"
-require "support/primary_indexing_operation_support"
+require "elastic_graph/spec_support/builds_indexer_operation"
 require "json"
 
 module ElasticGraph
   class Indexer
     RSpec.describe Processor do
       describe ".process", :factories, :capture_logs do
-        include PrimaryIndexingOperationSupport
+        include SpecSupport::BuildsIndexerOperation
 
         let(:clock) { class_double(Time, now: Time.iso8601("2020-09-15T12:30:00Z")) }
         let(:datastore_router) { instance_spy(ElasticGraph::Indexer::DatastoreIndexingRouter) }

@@ -4,7 +4,7 @@ Write ElasticGraph-shaped JSONL files to S3, packaged for AWS Lambda.
 
 This gem adapts ElasticGraph's indexing pipeline so that, instead of writing to the datastore,
 it writes batched, gzipped [JSON Lines](https://jsonlines.org/) (JSONL) files to Amazon S3. Each line in the file
-conforms to your ElasticGraph schema's latest JSON Schema for the corresponding object type.
+conforms to a specific JSON Schema version for the corresponding object type, with files partitioned by schema version.
 
 **Note:** This code does not deduplicate when writing to S3, so the data will contain all events
 and versions published, plus any Lambda retries. Consumers of the S3 bucket are responsible for
@@ -37,9 +37,9 @@ graph LR;
 
 ## What it does
 
-- Consumes ElasticGraph indexing operations and groups them by GraphQL type
+- Consumes ElasticGraph indexing operations and groups them by GraphQL type and JSON schema version
 - Transforms each operation into a flattened JSON document that matches your ElasticGraph schema
-- Writes one gzipped JSONL file per type per batch to S3 with deterministic keys:
+- Writes one gzipped JSONL file per type per JSON schema version per batch to S3 with deterministic keys:
   - `s3://<bucket>/<s3_path_prefix>/<TypeName>/v<json_schema_version>/<YYYY-MM-DD>/<uuid>.jsonl.gz`
 - Emits structured logs for observability (counts, sizes, S3 key, etc.)
 

@@ -0,0 +1,113 @@
+# 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/lambda_support"
+require "elastic_graph/support/from_yaml_file"
+require "elastic_graph/warehouse_lambda/config"
+
+module ElasticGraph
+  # Wraps an {Indexer} to dump data to S3 instead of indexing to a datastore.
+  # This is a stateful wrapper class (unlike {IndexerLambda} and {GraphQLLambda},
+  # which are namespace modules), as it manages the relationship between the
+  # indexer, S3 client, and warehouse dumper.
+  #
+  # @private
+  class WarehouseLambda
+    extend Support::FromYamlFile
+
+    # @return [Config] warehouse configuration
+    # @return [Indexer::Config] indexer configuration
+    # @return [DatastoreCore] datastore core for accessing schema artifacts
+    # @return [Logger] logger instance from datastore core
+    # @return [Module] clock module for time generation
+    # @dynamic config, indexer_config, datastore_core, logger, clock, indexer
+    attr_reader :config, :indexer_config, :datastore_core, :logger, :clock
+
+    # Builds an `ElasticGraph::WarehouseLambda` instance from parsed YAML configuration.
+    #
+    # @param parsed_yaml [Hash] parsed YAML configuration
+    # @yield [Datastore::Client] optional block to customize the datastore client
+    # @return [WarehouseLambda] configured warehouse lambda instance
+    def self.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block)
+      new(
+        config: Config.from_parsed_yaml!(parsed_yaml),
+        indexer_config: Indexer::Config.from_parsed_yaml(parsed_yaml) || Indexer::Config.new,
+        datastore_core: DatastoreCore.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block)
+      )
+    end
+
+    # Initializes a WarehouseLambda instance.
+    #
+    # @param config [Config] warehouse configuration
+    # @param indexer_config [Indexer::Config] indexer configuration
+    # @param datastore_core [DatastoreCore] datastore core for accessing schema artifacts
+    # @param clock [Module] clock module for time generation (defaults to {::Time})
+    # @param s3_client [Aws::S3::Client, nil] optional S3 client (for testing)
+    def initialize(config:, indexer_config:, datastore_core:, clock: ::Time, s3_client: nil)
+      @config = config
+      @indexer_config = indexer_config
+      @datastore_core = datastore_core
+      @logger = datastore_core.logger
+      @clock = clock
+      @s3_client = s3_client
+    end
+
+    # Returns the processor from the indexer for event processing.
+    #
+    # @return [Processor] the processor that handles incoming events
+    def processor
+      indexer.processor
+    end
+
+    # Returns the indexer instance, lazily building it on first access.
+    #
+    # @return [Indexer] the indexer that processes events
+    def indexer
+      @indexer ||= begin
+        require "elastic_graph/indexer"
+        Indexer.new(
+          config: indexer_config,
+          datastore_core: datastore_core,
+          datastore_router: warehouse_dumper,
+          clock: clock
+        )
+      end
+    end
+
+    # Returns the warehouse dumper instance, lazily building it on first access.
+    #
+    # @return [WarehouseDumper] the dumper that writes data to S3
+    def warehouse_dumper
+      @warehouse_dumper ||= begin
+        require "elastic_graph/warehouse_lambda/warehouse_dumper"
+        WarehouseDumper.new(
+          logger: logger,
+          s3_client: s3_client,
+          s3_bucket_name: config.s3_bucket_name,
+          s3_file_prefix: config.s3_path_prefix,
+          clock: clock
+        )
+      end
+    end
+
+    # Returns the S3 client instance, lazily building it on first access.
+    #
+    # @return [Aws::S3::Client] the S3 client for uploading data
+    def s3_client
+      @s3_client ||= begin
+        require "aws-sdk-s3"
+
+        if (region = config.aws_region)
+          ::Aws::S3::Client.new(region: region)
+        else
+          ::Aws::S3::Client.new
+        end
+      end
+    end
+  end
+end