unique_keys_tracker.rb

# frozen_string_literal: true

module SplitIoClient
  module Engine
    module Impressions
      class UniqueKeysTracker
        INTERVAL_TO_CLEAR_LONG_TERM_CACHE = 86_400 # 24 hours

        def initialize(config,
                       filter_adapter,
                       sender_adapter,
                       cache)
          @config = config
          @filter_adapter = filter_adapter
          @sender_adapter = sender_adapter
          @cache = cache
          @max_bulk_size = config.unique_keys_bulk_size
          @semaphore = Mutex.new
          @keys_size = 0
        end

        def call
          @config.threads[:unique_keys_sender] = Thread.new { send_bulk_data_thread }
          @config.threads[:clear_filter] = Thread.new { clear_filter_thread }
        end

        def track(feature_name, key)
          return false if @filter_adapter.contains?(feature_name, key)

          @filter_adapter.add(feature_name, key)

          add_or_update(feature_name, key)
          @keys_size += 1

          send_bulk_data if @keys_size >= @max_bulk_size

          true
        rescue StandardError => e
          @config.log_found_exception(__method__.to_s, e)
          false
        end

        private

        def send_bulk_data_thread
          @config.logger.info('Starting Unique Keys Tracker.') if @config.debug_enabled
          loop do
            sleep(@config.unique_keys_refresh_rate)
            send_bulk_data
          end
        rescue SplitIoClient::SDKShutdownException
          send_bulk_data
          @config.logger.info('Posting unique keys due to shutdown')
        end

        def clear_filter_thread
          loop do
            sleep(INTERVAL_TO_CLEAR_LONG_TERM_CACHE)
            @config.logger.debug('Starting task to clean the filter cache.') if @config.debug_enabled
            @filter_adapter.clear
          end
        rescue SplitIoClient::SDKShutdownException
          @filter_adapter.clear
        end

        def add_or_update(feature_name, key)
          if @cache[feature_name].nil?
            @cache[feature_name] = Set.new([key])
          else
            @cache[feature_name].add(key)
          end
        end

        def clear_cache
          uniques = @cache.clone
          keys_size = @keys_size
          @cache.clear
          @keys_size = 0

          [uniques, keys_size]
        end

        def send_bulk_data
          @semaphore.synchronize do
            return if @cache.empty?

            uniques, keys_size = clear_cache
            if keys_size <= @max_bulk_size
              @sender_adapter.record_uniques_key(uniques)
              return

            end
            bulks = flatten_bulks(uniques)
            bulks_to_post = group_bulks_by_max_size(bulks)
            @sender_adapter.record_uniques_key(bulks_to_post)
          end
        rescue StandardError => e
          @config.log_found_exception(__method__.to_s, e)
        end

        def group_bulks_by_max_size(bulks)
          current_size = 0
          bulks_to_post = Concurrent::Hash.new
          bulks.each do |bulk|
            key, value = bulk.first
            if (value.size + current_size) > @max_bulk_size
              @sender_adapter.record_uniques_key(bulks_to_post)
              bulks_to_post = Concurrent::Hash.new
              current_size = 0
            end
            bulks_to_post[key] = value
            current_size += value.size
          end

          bulks_to_post
        end

        def flatten_bulks(uniques)
          bulks = []
          uniques.each_key do |unique_key|
            bulks += check_keys_and_split_to_bulks(uniques[unique_key], unique_key)
          end

          bulks
        end

        def check_keys_and_split_to_bulks(value, key)
          unique_updated = []
          if value.size > @max_bulk_size
            sub_bulks = SplitIoClient::Utilities.split_bulk_to_send(value, @max_bulk_size)
            sub_bulks.each do |sub_bulk|
              unique_updated << { key => sub_bulk.to_set }
            end
            return unique_updated

          end
          unique_updated << { key => value }

          unique_updated
        end
      end
    end
  end
end