Refactor Data Streams configuration to follow AppSec pattern

- Move Data Streams configuration to use self.extended hook pattern
- Add Extensions module to activate Data Streams configuration
- Move ENV_ENABLED constant to lib/datadog/data_streams/ext.rb
- Add comprehensive RBS type signatures for Data Streams modules
- Fix type errors: add missing timestamp field to consumer stats
- Improve type coverage to 88-100% across data_streams files
- Add configuration settings tests

This addresses review comments to align with the existing AppSec
configuration pattern and improve type safety.

Author: ericfirth

lib/datadog/core/configuration/settings.rb

@@ -12,7 +12,6 @@
 require_relative '../../profiling/ext'
 
 require_relative '../../tracing/configuration/settings'
-require_relative '../../data_streams/configuration/settings'
 
 module Datadog
   module Core
@@ -1031,7 +1030,6 @@ def initialize(*_)
         # TODO: Tracing should manage its own settings.
         #       Keep this extension here for now to keep things working.
         extend Datadog::Tracing::Configuration::Settings
-        extend Datadog::DataStreams::Configuration::Settings
       end
       # standard:enable Metrics/BlockLength
     end

lib/datadog/data_streams.rb

@@ -3,6 +3,7 @@
 require_relative 'data_streams/processor'
 require_relative 'data_streams/pathway_context'
 require_relative 'data_streams/configuration/settings'
+require_relative 'data_streams/extensions'
 require_relative 'core/utils/time'
 
 module Datadog
@@ -104,5 +105,8 @@ def components
         Datadog.send(:components)
       end
     end
+
+    # Expose Data Streams to global shared objects
+    Extensions.activate!
   end
 end

lib/datadog/data_streams/configuration.rb

@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+
+require_relative 'configuration/settings'
+
+module Datadog
+  module DataStreams
+    # Configuration for Data Streams Monitoring
+    module Configuration
+    end
+  end
+end
+

lib/datadog/data_streams/configuration/settings.rb

@@ -1,13 +1,19 @@
 # frozen_string_literal: true
 
 require_relative '../../core/environment/variable_helpers'
+require_relative '../ext'
 
 module Datadog
   module DataStreams
     module Configuration
       # Configuration settings for Data Streams Monitoring.
       module Settings
         def self.extended(base)
+          base = base.singleton_class unless base.is_a?(Class)
+          add_settings!(base)
+        end
+
+        def self.add_settings!(base)
           base.class_eval do
             # Data Streams Monitoring configuration
             # @public_api
@@ -19,7 +25,7 @@ def self.extended(base)
               # @return [Boolean]
               option :enabled do |o|
                 o.type :bool
-                o.env 'DD_DATA_STREAMS_ENABLED'
+                o.env Ext::ENV_ENABLED
                 o.default false
               end
 

lib/datadog/data_streams/ext.rb

@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+
+module Datadog
+  module DataStreams
+    # Constants for Data Streams Monitoring configuration
+    # @public_api
+    module Ext
+      ENV_ENABLED = 'DD_DATA_STREAMS_ENABLED'
+    end
+  end
+end
+

lib/datadog/data_streams/extensions.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+require_relative 'configuration'
+require_relative '../core/configuration'
+
+module Datadog
+  module DataStreams
+    # Extends Datadog with Data Streams Monitoring features
+    module Extensions
+      # Inject Data Streams settings into global configuration.
+      def self.activate!
+        Core::Configuration::Settings.extend(Configuration::Settings)
+      end
+    end
+  end
+end
+

lib/datadog/data_streams/processor.rb

@@ -246,7 +246,7 @@ def decode_pathway_b64(encoded_ctx)
       end
 
       def flush_stats
-        payload = nil
+        payload = nil # : ::Hash[::String, untyped]?
 
         @stats_mutex.synchronize do
           return if @buckets.empty? && @consumer_stats.empty?
@@ -357,6 +357,7 @@ def record_consumer_stats(topic:, partition:, offset:, timestamp:)
             topic: topic,
             partition: partition,
             offset: offset,
+            timestamp: timestamp,
             timestamp_sec: timestamp.to_f
           }
 
@@ -405,7 +406,7 @@ def aggregate_stats_by_time_buckets
 
       def send_stats_to_agent(payload)
         response = transport.send_stats(payload)
-        @logger.debug("DSM stats sent to agent: #{response.code if response.respond_to?(:code)}")
+        @logger.debug("DSM stats sent to agent: ok=#{response.ok?}")
       end
 
       def transport

lib/datadog/tracing/configuration/ext.rb

@@ -110,11 +110,6 @@ module HTTPErrorStatuses
           ENV_SERVER_ERROR_STATUSES = 'DD_TRACE_HTTP_SERVER_ERROR_STATUSES'
           ENV_CLIENT_ERROR_STATUSES = 'DD_TRACE_HTTP_CLIENT_ERROR_STATUSES'
         end
-
-        # @public_api
-        module DataStreams
-          ENV_ENABLED = 'DD_DATA_STREAMS_ENABLED'
-        end
       end
     end
   end

sig/datadog/data_streams/configuration/settings.rbs

@@ -2,7 +2,7 @@ module Datadog
   module DataStreams
     module Configuration
       module Settings
-        def self.extended: (untyped base) -> void
+        def self.extended: (Module base) -> void
       end
     end
   end

sig/datadog/data_streams/ext.rbs

@@ -0,0 +1,8 @@
+module Datadog
+  module DataStreams
+    module Ext
+      ENV_ENABLED: ::String
+    end
+  end
+end
+