feat: Support getExperimentFlag, trackEvent and trackExposure methods

.github/workflows/sdk-spec.yml

@@ -11,8 +11,8 @@ permissions:
 jobs:
   sdk-spec:
     # Keep the ref in `uses:` and `spec-ref` in sync.
-    uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.0
+    uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.1
     with:
-      spec-ref: v0.7.0
-      tags: core or local-evaluation
+      spec-ref: v0.7.1
+      tags: core or events or local-evaluation
       language: ruby

lib/flagsmith.rb

@@ -14,6 +14,7 @@
 require 'flagsmith/sdk/config'
 require 'flagsmith/sdk/errors'
 require 'flagsmith/sdk/intervals'
+require 'flagsmith/sdk/event_processor'
 require 'flagsmith/sdk/utils'
 require 'flagsmith/sdk/pooling_manager'
 require 'flagsmith/sdk/models/flags'
@@ -44,13 +45,18 @@ class Client # rubocop:disable Metrics/ClassLength
     # feature_value_for_identity = identity_flags.get_feature_value('foo')
     #
     # identity_segments = flagsmith.get_identity_segments('identifier', {'foo': 'bar'})
+    #
+    # With enable_events: true, experiment flags record exposures, and events can be tracked:
+    #
+    # flag = flagsmith.get_experiment_flag('foo', 'identifier', plan: 'premium')
+    # flagsmith.track_event('purchase', identifier: 'identifier', value: 42)
 
     # Available Configs.
     #
     # :environment_key, :api_url, :custom_headers, :request_timeout_seconds, :enable_local_evaluation,
     # :environment_refresh_interval_seconds, :retries, :enable_analytics, :default_flag_handler,
     # :offline_mode, :offline_handler, :polling_manager_failure_limit
-    # :realtime_api_url, :enable_realtime_updates, :logger
+    # :realtime_api_url, :enable_realtime_updates, :logger, :enable_events, :events_api_url
     #
     # You can see full description in the Flagsmith::Config
 
@@ -68,6 +74,7 @@ def initialize(config)
 
       api_client
       analytics_processor
+      event_processor
       environment_data_polling_manager
       load_offline_handler
     end
@@ -77,6 +84,7 @@ def validate_offline_mode!
         raise Flagsmith::ClientError,
               'The offline_mode config param requires a matching offline_handler.'
       end
+      raise Flagsmith::ClientError, 'The enable_events config param cannot be used with offline_mode.' if @config.offline_mode? && @config.enable_events?
       return unless @config.offline_handler && @config.default_flag_handler
 
       raise Flagsmith::ClientError,
@@ -109,6 +117,12 @@ def analytics_processor
         )
     end
 
+    def event_processor
+      return nil unless @config.enable_events?
+
+      @event_processor ||= Flagsmith::EventProcessor.new(@config).tap(&:start)
+    end
+
     def load_offline_handler
       @environment = offline_handler.environment if offline_handler
     end
@@ -205,6 +219,47 @@ def get_value_for_identity(feature_name, user_id = nil, default: nil)
       flag.value
     end
 
+    # Get a flag for an identity, taking the same traits as get_identity_flags, and record the
+    # identity's exposure to the variant it was served when it is in the feature's running experiment.
+    # Requires enable_events. Exposure needs remote evaluation: the environment document used by local
+    # evaluation and offline handlers carries no experiment data, so no exposure is recorded from it.
+    def get_experiment_flag(feature_name, identifier, **traits)
+      require_events!
+      flag = get_identity_flags(identifier, **traits)[feature_name]
+      skip_reason = exposure_skip_reason(flag)
+      if skip_reason
+        @config.logger.debug "Not sending #{Flagsmith::EventProcessor::FLAG_EXPOSURE_EVENT} for feature #{feature_name}: #{skip_reason}."
+      else
+        track_exposure_event(feature_name, identifier: identifier, value: flag.variant, traits: traits,
+                                           metadata: { experiment_id: flag.experiment.id })
+      end
+      flag
+    end
+
+    # Send an application event, e.g. a conversion, to Flagsmith. Requires enable_events.
+    def track_event(event, identifier: nil, value: nil, traits: nil, metadata: nil)
+      require_events!
+      if event.start_with?('$')
+        raise Flagsmith::ClientError,
+              "Event name \"#{event}\" uses the \"$\" prefix, which is reserved for events sent by the SDK."
+      end
+
+      event_processor.track_event(event, identifier: identifier, value: value, traits: traits, metadata: metadata)
+    end
+
+    # Record that an identity was shown a variant of a feature. Requires enable_events.
+    def track_exposure_event(feature_name, identifier:, value: nil, traits: nil, metadata: nil)
+      require_events!
+      if identifier.to_s.empty?
+        @config.logger.warn "Not sending #{Flagsmith::EventProcessor::FLAG_EXPOSURE_EVENT} for feature #{feature_name}: " \
+                            'an exposure requires an identifier to reconcile with conversion events.'
+        return
+      end
+
+      event_processor.track_exposure_event(feature_name, identifier: identifier, value: value, traits: traits,
+                                                         metadata: metadata)
+    end
+
     def get_identity_segments(identifier, traits = {})
       raise Flagsmith::ClientError, 'Local evaluation or offline handler is required to obtain identity segments.' unless environment
 
@@ -221,6 +276,19 @@ def get_identity_segments(identifier, traits = {})
 
     private
 
+    def require_events!
+      raise Flagsmith::ClientError, 'The enable_events config param is required to send events.' unless event_processor
+    end
+
+    def exposure_skip_reason(flag)
+      return 'feature not found' unless flag.is_a?(Flagsmith::Flags::Flag)
+      return 'feature is disabled' unless flag.enabled?
+      return 'flags come from the environment document, which carries no experiment data' if environment
+      return 'identity is not in a running experiment' unless flag.experiment&.in_experiment?
+
+      'flag has no variant' if flag.variant.nil?
+    end
+
     def environment_flags_from_document # rubocop:disable Metrics/MethodLength
       unless environment
         raise Flagsmith::ClientError,

lib/flagsmith/sdk/config.rb

@@ -5,12 +5,13 @@ module Flagsmith
   class Config
     DEFAULT_API_URL = 'https://edge.api.flagsmith.com/api/v1/'
     DEFAULT_REALTIME_API_URL = 'https://realtime.flagsmith.com/'
+    DEFAULT_EVENTS_API_URL = 'https://events.api.flagsmith.com/'
 
     OPTIONS = %i[
       environment_key api_url custom_headers request_timeout_seconds enable_local_evaluation
       environment_refresh_interval_seconds retries enable_analytics default_flag_handler
       offline_mode offline_handler polling_manager_failure_limit
-      realtime_api_url enable_realtime_updates logger
+      realtime_api_url enable_realtime_updates logger enable_events events_api_url
     ].freeze
 
     # Available Configs
@@ -47,6 +48,9 @@ class Config
     #                                            non-standard realtime endpoint.
     #   +enable_realtime_updates+              - A boolean to enable realtime updates.
     #   +logger+                               - Pass your logger, default is Logger.new($stdout)
+    #   +enable_events+                        - if enabled, sends tracking and exposure events to the
+    #                                            Flagsmith events API to power experiments
+    #   +events_api_url+                       - Override the URL of the Flagsmith events API
     #
     attr_reader(*OPTIONS)
 
@@ -72,6 +76,10 @@ def realtime_mode?
       @enable_realtime_updates
     end
 
+    def enable_events?
+      @enable_events
+    end
+
     def environment_flags_url
       'flags/'
     end
@@ -106,9 +114,15 @@ def build_config(options)
       @realtime_api_url << '/' unless @realtime_api_url.end_with? '/'
       @enable_realtime_updates = opts.fetch(:enable_realtime_updates, false)
       @logger = options.fetch(:logger, Logger.new($stdout).tap { |l| l.level = :debug })
+      @enable_events = opts.fetch(:enable_events, false)
+      @events_api_url = with_trailing_slash(opts.fetch(:events_api_url, Flagsmith::Config::DEFAULT_EVENTS_API_URL))
     end
     # rubocop:enable Metrics/AbcSize, Metrics/MethodLength
 
+    def with_trailing_slash(url)
+      url.end_with?('/') ? url : "#{url}/"
+    end
+
     class << self
       def environment_key
         ENV.fetch('FLAGSMITH_ENVIRONMENT_KEY', nil)

lib/flagsmith/sdk/event_processor.rb

@@ -0,0 +1,163 @@
+# frozen_string_literal: true
+
+require 'set'
+
+module Flagsmith
+  # Started event processors, stopped together at exit. A stopped processor leaves, so it is not kept in memory.
+  module RunningEventProcessors
+    @processors = Set.new
+    @mutex = Mutex.new
+    at_exit { to_a.each(&:stop) }
+
+    def self.add(processor)
+      @mutex.synchronize { @processors.add(processor) }
+    end
+
+    def self.delete(processor)
+      @mutex.synchronize { @processors.delete(processor) }
+    end
+
+    def self.to_a
+      @mutex.synchronize { @processors.to_a }
+    end
+  end
+
+  # Buffers tracking and exposure events, and sends them to the Flagsmith events API in batches.
+  #
+  # Events are sent every +flush_interval_seconds+, as soon as +max_buffer_items+ are buffered,
+  # or on #flush. Sending events never raises into the application: failures are logged.
+  class EventProcessor # rubocop:disable Metrics/ClassLength
+    include Flagsmith::SDK::Intervals
+
+    FLAG_EXPOSURE_EVENT = '$flag_exposure'
+    ENDPOINT = 'v1/events'
+    FLUSH_INTERVAL_SECONDS = 10
+    MAX_BUFFER_ITEMS = 1000
+    REQUEST_TIMEOUT_SECONDS = 3
+    RETRIABLE_STATUSES = [500, 502, 503, 504].freeze
+
+    def initialize(config, flush_interval_seconds: FLUSH_INTERVAL_SECONDS, max_buffer_items: MAX_BUFFER_ITEMS)
+      @logger = config.logger
+      @flush_interval_seconds = flush_interval_seconds
+      @max_buffer_items = max_buffer_items
+      @connection = build_connection(config)
+      @events = []
+      @buffered_exposures = Set.new
+      @mutex = Mutex.new
+      @flush_mutex = Mutex.new
+      @flush_pending = false
+    end
+
+    def track_event(event, identifier: nil, value: nil, traits: nil, metadata: nil)
+      buffer(payload(event, nil, identifier: identifier, value: value, traits: traits, metadata: metadata))
+    end
+
+    # Equal exposures within one flush interval add no information, so only the first is sent.
+    def track_exposure_event(feature_name, identifier:, value: nil, traits: nil, metadata: nil)
+      buffer(payload(FLAG_EXPOSURE_EVENT, feature_name,
+                     identifier: identifier, value: value, traits: traits, metadata: metadata))
+    end
+
+    def flush
+      drain
+    end
+
+    def start
+      return if @interval
+
+      @interval = set_interval(@flush_interval_seconds) { flush }
+      RunningEventProcessors.add(self)
+    end
+
+    def stop
+      @flush_mutex.synchronize { clear_interval(@interval) } if @interval # never interrupts a batch being sent
+      @interval = nil
+      RunningEventProcessors.delete(self)
+      flush
+    end
+
+    private
+
+    # Only the flush scheduled by a full buffer clears +@flush_pending+, once it holds the buffer, so at most one
+    # such flush is waiting at a time.
+    def scheduled_flush
+      drain { @flush_pending = false }
+    end
+
+    def drain
+      @flush_mutex.synchronize do
+        batch = @mutex.synchronize do
+          buffered = @events
+          @events = []
+          @buffered_exposures.clear
+          yield if block_given?
+          buffered
+        end
+        send_events(batch) unless batch.empty?
+      end
+    end
+
+    def build_connection(config)
+      Faraday.new(url: config.events_api_url, headers: headers(config)) do |f|
+        f.options.timeout = REQUEST_TIMEOUT_SECONDS
+        f.response :raise_error
+        f.request :retry, max: 1, interval: 1, methods: %i[post], retry_statuses: RETRIABLE_STATUSES,
+                          exceptions: [*Faraday::Retry::Middleware::DEFAULT_EXCEPTIONS, Faraday::ConnectionFailed]
+      end
+    end
+
+    def headers(config)
+      config.custom_headers.merge(
+        'Content-Type' => 'application/json; charset=utf-8',
+        'X-Environment-Key' => config.environment_key,
+        'User-Agent' => Flagsmith::SDK::Utils.user_agent,
+        'Flagsmith-SDK-User-Agent' => Flagsmith::SDK::Utils.user_agent
+      )
+    end
+
+    def payload(event, feature_name, details)
+      {
+        event: event,
+        feature_name: feature_name,
+        identifier: details[:identifier],
+        value: details[:value]&.to_s,
+        traits: trait_values(details[:traits]),
+        metadata: (details[:metadata] || {}).transform_keys(&:to_s).merge('sdk_version' => Flagsmith::VERSION),
+        timestamp: (Time.now.to_f * 1000).to_i
+      }
+    end
+
+    # Traits given as { value:, transient: } configurations are sent as their value.
+    def trait_values(traits)
+      return nil if traits.nil? || traits.empty?
+
+      traits.to_h { |key, value| [key.to_s, value.is_a?(Hash) ? value[:value] : value] }
+    end
+
+    def buffer(event)
+      schedule_flush = @mutex.synchronize do
+        next false if duplicate_exposure?(event)
+
+        @events << event
+        next false if @flush_pending || @events.length < @max_buffer_items
+
+        @flush_pending = true
+      end
+      Thread.new { scheduled_flush } if schedule_flush
+    end
+
+    def duplicate_exposure?(event)
+      return false unless event[:event] == FLAG_EXPOSURE_EVENT
+      return false if @buffered_exposures.add?(event.values_at(:feature_name, :identifier, :value))
+
+      @logger.debug "Skipping duplicate #{FLAG_EXPOSURE_EVENT} event for feature #{event[:feature_name]}"
+      true
+    end
+
+    def send_events(events)
+      @connection.post(ENDPOINT, { events: events }.to_json)
+    rescue StandardError => e
+      @logger.warn "Unable to send #{events.length} events to the Flagsmith events API, dropping them: #{e}"
+    end
+  end
+end