<commit_before>require 'ostruct'
require 'thread'

module RubyEventStore
  class InMemoryRepository

    def initialize
      @all = Array.new
      @streams = Hash.new
      @mutex = Mutex.new
    end

    def append_to_stream(events, stream_name, expected_version)
      add_to_stream(events, expected_version, stream_name, true)
    end

    def link_to_stream(event_ids, stream_name, expected_version)
      events = normalize_to_array(event_ids).map{|eid| read_event(eid) }
      add_to_stream(events, expected_version, stream_name, nil)
    end

    def delete_stream(stream_name)
      @streams.delete(stream_name)
    end

    def has_event?(event_id)
      @all.any?{ |item| item.event_id.eql?(event_id) }
    end

    def last_stream_event(stream_name)
      read_stream_events_forward(stream_name).last
    end

    def read_events_forward(stream_name, start_event_id, count)
      source = read_stream_events_forward(stream_name)
      read_batch(source, start_event_id, count)
    end

    def read_events_backward(stream_name, start_event_id, count)
      source = read_stream_events_backward(stream_name)
      read_batch(source, start_event_id, count)
    end

    def read_stream_events_forward(stream_name)
      return @all if stream_name.eql?(GLOBAL_STREAM)
      @streams[stream_name] || Array.new
    end

    def read_stream_events_backward(stream_name)
      return @all if stream_name.eql?(GLOBAL_STREAM)
      read_stream_events_forward(stream_name).reverse
    end

    def read_all_streams_forward(start_event_id, count)
      read_batch(@all, start_event_id, count)
    end

    def read_all_streams_backward(start_event_id, count)
      read_batch(@all.reverse, start_event_id, count)
    end

    def read_event(event_id)
      @all.find { |e| event_id.eql?(e.event_id) } or raise EventNotFound.new(event_id)
    end

    def get_all_streams
      [Stream.new("all")] + @streams.keys.map { |name| Stream.new(name) }
    end

    private

    def normalize_to_array(events)
      return *events
    end

    def add_to_stream(events, expected_version, stream_name, include_global)
      raise InvalidExpectedVersion if !expected_version.equal?(:any) && stream_name.eql?(GLOBAL_STREAM)
      events = normalize_to_array(events)
      expected_version = case expected_version
        when :none
          -1
        when :auto
          read_stream_events_forward(stream_name).size - 1
        when Integer, :any
          expected_version
        else
         raise InvalidExpectedVersion
      end
      append_with_synchronize(events, expected_version, stream_name, include_global)
    end

    def append_with_synchronize(events, expected_version, stream_name, include_global)
      # expected_version :auto assumes external lock is used
      # which makes reading stream before writing safe.
      #
      # To emulate potential concurrency issues of :auto strategy without
      # such external lock we use Thread.pass to make race
      # conditions more likely. And we only use mutex.synchronize for writing
      # not for the whole read+write algorithm.
      Thread.pass
      @mutex.synchronize do
        if expected_version == :any
          expected_version = read_stream_events_forward(stream_name).size - 1
        end
        append(events, expected_version, stream_name, include_global)
      end
    end

    def append(events, expected_version, stream_name, include_global)
      stream = read_stream_events_forward(stream_name)
      raise WrongExpectedEventVersion unless (stream.size - 1).equal?(expected_version)
      events.each do |event|
        raise EventDuplicatedInStream if stream.any?{|ev| ev.event_id.eql?(event.event_id) }
        if include_global
          raise EventDuplicatedInStream if @all.any?{|ev| ev.event_id.eql?(event.event_id) }
          @all.push(event)
        end
        stream.push(event) unless stream_name.eql?(GLOBAL_STREAM)
      end
      @streams[stream_name] = stream
      self
    end<commit_msg>Refactor: GLOBAL_STREAM like any other stream.<commit_after>require 'ostruct'
require 'thread'

module RubyEventStore
  class InMemoryRepository

    def initialize
      @streams = Hash.new
      @mutex = Mutex.new
      @streams[GLOBAL_STREAM] = Array.new
    end

    def append_to_stream(events, stream_name, expected_version)
      add_to_stream(events, expected_version, stream_name, true)
    end

    def link_to_stream(event_ids, stream_name, expected_version)
      events = normalize_to_array(event_ids).map{|eid| read_event(eid) }
      add_to_stream(events, expected_version, stream_name, nil)
    end

    def delete_stream(stream_name)
      @streams.delete(stream_name)
    end

    def has_event?(event_id)
      @streams.fetch(GLOBAL_STREAM).any? { |item| item.event_id.eql?(event_id) }
    end

    def last_stream_event(stream_name)
      read_stream_events_forward(stream_name).last
    end

    def read_events_forward(stream_name, start_event_id, count)
      source = read_stream_events_forward(stream_name)
      read_batch(source, start_event_id, count)
    end

    def read_events_backward(stream_name, start_event_id, count)
      source = read_stream_events_backward(stream_name)
      read_batch(source, start_event_id, count)
    end

    def read_stream_events_forward(stream_name)
      @streams.fetch(stream_name, Array.new)
    end

    def read_stream_events_backward(stream_name)
      read_stream_events_forward(stream_name).reverse
    end

    def read_all_streams_forward(start_event_id, count)
      read_events_forward(GLOBAL_STREAM, start_event_id, count)
    end

    def read_all_streams_backward(start_event_id, count)
      read_events_backward(GLOBAL_STREAM, start_event_id, count)
    end

    def read_event(event_id)
      @streams.fetch(GLOBAL_STREAM).find { |e| event_id.eql?(e.event_id) } or raise EventNotFound.new(event_id)
    end

    def get_all_streams
      @streams.keys.map { |name| Stream.new(name) }
    end

    private

    def normalize_to_array(events)
      return *events
    end

    def add_to_stream(events, expected_version, stream_name, include_global)
      raise InvalidExpectedVersion if !expected_version.equal?(:any) && stream_name.eql?(GLOBAL_STREAM)
      events = normalize_to_array(events)
      expected_version = case expected_version
        when :none
          -1
        when :auto
          read_stream_events_forward(stream_name).size - 1
        when Integer, :any
          expected_version
        else
         raise InvalidExpectedVersion
      end
      append_with_synchronize(events, expected_version, stream_name, include_global)
    end

    def append_with_synchronize(events, expected_version, stream_name, include_global)
      # expected_version :auto assumes external lock is used
      # which makes reading stream before writing safe.
      #
      # To emulate potential concurrency issues of :auto strategy without
      # such external lock we use Thread.pass to make race
      # conditions more likely. And we only use mutex.synchronize for writing
      # not for the whole read+write algorithm.
      Thread.pass
      @mutex.synchronize do
        if expected_version == :any
          expected_version = read_stream_events_forward(stream_name).size - 1
        end
        append(events, expected_version, stream_name, include_global)
      end
    end

    def append(events, expected_version, stream_name, include_global)
      stream = read_stream_events_forward(stream_name)
      raise WrongExpectedEventVersion unless (stream.size - 1).equal?(expected_version)
      events.each do |event|
        raise EventDuplicatedInStream if stream.any?{|ev| ev.event_id.eql?(event.event_id) }
        if include_global
          global_stream = read_stream_events_forward(GLOBAL_STREAM)
          raise EventDuplicatedInStream if global_stream.any? { |ev| ev.event_id.eql?(event.event_id) }
          global_stream.push(event)
        end
        stream.push(event) unless stream_name.eql?(GLOBAL_STREAM)
      end
      @streams[stream_name] = stream
      self
    end
{'fraction_non_alphanumeric': 0.07279029462738301, 'fraction_numerical': 0.0009903441445902451, 'mean_word_length': 3.3272629887520084, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4262828', 'n_tokens_mistral': 2488, 'n_tokens_neox': 2461, 'n_words': 530}