Add AWS Lambda integration with LambdaFunction (#975)

* Add Lambda handler for elasticgraph-warehouse_lambda

This commit completes the AWS Lambda integration for elasticgraph-warehouse_lambda
by adding the Lambda function handler that serves as the deployment entry point.

Key components:
- LambdaFunction class that wraps WarehouseLambda with AWS Lambda lifecycle
- DumpWarehouseData constant exposed as the Lambda handler entry point
- Integration with IndexerLambda::SqsProcessor for SQS event processing
- Support for IGNORE_SQS_LATENCY_TIMESTAMPS_FROM_ARNS environment variable
- warehouse_lambda_from_env class method for building from Lambda ENV vars
- Full RBS type signatures for type safety
- 100% test coverage with comprehensive behavioral tests

Configuration:
- Reuses SQS event format from elasticgraph-indexer_lambda
- Processes batches of indexing operations and writes JSONL to S3
- Returns batch item failures for SQS partial batch failure handling

Documentation:
- Updated README with Lambda handler example showing DumpWarehouseData usage
- Documented deployment pattern for AWS Lambda

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Myron Marston <myron.marston@gmail.com>

Author: 3 people

elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb

@@ -6,6 +6,8 @@
 #
 # frozen_string_literal: true
 
+require "elastic_graph/datastore_core"
+require "elastic_graph/indexer/config"
 require "elastic_graph/lambda_support"
 require "elastic_graph/support/from_yaml_file"
 require "elastic_graph/warehouse_lambda/config"
@@ -20,6 +22,11 @@ module ElasticGraph
   class WarehouseLambda
     extend Support::FromYamlFile
 
+    # Builds an `ElasticGraph::WarehouseLambda` instance from our lambda ENV vars.
+    def self.warehouse_lambda_from_env
+      LambdaSupport.build_from_env(WarehouseLambda)
+    end
+
     # @return [Config] warehouse configuration
     # @return [Indexer::Config] indexer configuration
     # @return [DatastoreCore] datastore core for accessing schema artifacts
@@ -44,7 +51,7 @@ def self.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block)
     # Initializes a WarehouseLambda instance.
     #
     # @param config [Config] warehouse configuration
-    # @param indexer_config [Indexer::Config] indexer configuration
+    # @param indexer_config [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)

elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb

@@ -0,0 +1,43 @@
+# 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/lambda_function"
+require "json"
+
+module ElasticGraph
+  class WarehouseLambda
+    # @private
+    class LambdaFunction
+      prepend LambdaSupport::LambdaFunction
+
+      # @dynamic sqs_processor
+      attr_reader :sqs_processor
+
+      def initialize
+        require "elastic_graph/warehouse_lambda"
+        require "elastic_graph/indexer_lambda/sqs_processor"
+
+        warehouse_lambda = WarehouseLambda.warehouse_lambda_from_env
+        ignore_sqs_latency_timestamps_from_arns = ::JSON.parse(ENV.fetch("IGNORE_SQS_LATENCY_TIMESTAMPS_FROM_ARNS", "[]")).to_set
+
+        @sqs_processor = IndexerLambda::SqsProcessor.new(
+          warehouse_lambda.processor,
+          ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns,
+          logger: warehouse_lambda.logger
+        )
+      end
+
+      def handle_request(event:, context:)
+        sqs_processor.process(event)
+      end
+    end
+  end
+end
+
+# Lambda handler for `elasticgraph-warehouse_lambda`.
+DumpWarehouseData = ElasticGraph::WarehouseLambda::LambdaFunction.new

elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs

@@ -3,6 +3,8 @@ module ElasticGraph
     extend Support::FromYamlFile[WarehouseLambda]
     extend _BuildableFromParsedYaml[WarehouseLambda]
 
+    def self.warehouse_lambda_from_env: () -> WarehouseLambda
+
     attr_reader config: Config
     attr_reader indexer_config: Indexer::Config
     attr_reader datastore_core: DatastoreCore

elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/lambda_function.rbs

@@ -0,0 +1,19 @@
+# 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 WarehouseLambda
+    class LambdaFunction
+      include LambdaSupport::LambdaFunction[void]
+      include LambdaSupport::_LambdaFunctionClass[void]
+      attr_reader sqs_processor: IndexerLambda::SqsProcessor
+    end
+  end
+end
+
+DumpWarehouseData: ElasticGraph::WarehouseLambda::LambdaFunction

elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/lambda_function_spec.rb

@@ -0,0 +1,43 @@
+# 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/spec_support/lambda_function"
+
+RSpec.describe "Warehouse lambda function" do
+  include_context "lambda function", config_overrides_in_yaml: {"warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}}
+
+  it "defines DumpWarehouseData constant" do
+    expect_loading_lambda_to_define_constant(
+      lambda: "elastic_graph/warehouse_lambda/lambda_function.rb",
+      const: :DumpWarehouseData
+    ) do |lambda_function|
+      expect(lambda_function).to be_a(ElasticGraph::WarehouseLambda::LambdaFunction)
+      response = lambda_function.handle_request(event: {"Records" => []}, context: {})
+      expect(response).to eq({"batchItemFailures" => []})
+      expect(lambda_function.sqs_processor.ignore_sqs_latency_timestamps_from_arns).to eq([].to_set)
+    end
+  end
+
+  it "configures `ignore_sqs_latency_timestamps_from_arns` based on an ENV var" do
+    env_var_value = ::JSON.generate(["ignored-arn1", "ignored-arn2"])
+
+    with_env "IGNORE_SQS_LATENCY_TIMESTAMPS_FROM_ARNS" => env_var_value do
+      expect_loading_lambda_to_define_constant(
+        lambda: "elastic_graph/warehouse_lambda/lambda_function.rb",
+        const: :DumpWarehouseData
+      ) do |lambda_function|
+        response = lambda_function.handle_request(event: {"Records" => []}, context: {})
+        expect(response).to eq({"batchItemFailures" => []})
+        expect(lambda_function.sqs_processor.ignore_sqs_latency_timestamps_from_arns).to eq([
+          "ignored-arn1",
+          "ignored-arn2"
+        ].to_set)
+      end
+    end
+  end
+end