<commit_before>require 'activerecord-import'

module RailsEventStoreActiveRecord
  class EventRepository
    POSITION_SHIFT = 1

    def initialize
      verify_correct_schema_present
      @repo_reader = EventRepositoryReader.new
    end

    def append_to_stream(events, stream, expected_version)
      add_to_stream(normalize_to_array(events), stream, expected_version, true) do |event|
        build_event_record(event).save!
        event.event_id
      end
    end

    def link_to_stream(event_ids, stream, expected_version)
      (normalize_to_array(event_ids) - Event.where(id: event_ids).pluck(:id)).each do |id|
        raise RubyEventStore::EventNotFound.new(id)
      end
      add_to_stream(normalize_to_array(event_ids), stream, expected_version, nil) do |event_id|
        event_id
      end
    end

    def delete_stream(stream)
      EventInStream.where(stream: stream.name).delete_all
    end

    def has_event?(event_id)
      @repo_reader.has_event?(event_id)
    end

    def last_stream_event(stream)
      @repo_reader.last_stream_event(stream)
    end

    def read_events_forward(stream, after_event_id, count)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_events_forward(stream, after_event_id, count)
    end

    def read_events_backward(stream, before_event_id, count)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_events_backward(stream, before_event_id, count)
    end

    def read_stream_events_forward(stream)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_stream_events_forward(stream)
    end

    def read_stream_events_backward(stream)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_stream_events_backward(stream)
    end

    def read_all_streams_forward(after_event_id, count)
      @repo_reader.read_all_streams_forward(after_event_id, count)
    end

    def read_all_streams_backward(before_event_id, count)
      @repo_reader.read_all_streams_backward(before_event_id, count)
    end

    def read_event(event_id)
      @repo_reader.read_event(event_id)
    end

    def read(_)
      [].each
    end

    private

    def add_to_stream(collection, stream, expected_version, include_global, &to_event_id)
      last_stream_version = ->(stream_) { EventInStream.where(stream: stream_.name).order("position DESC").first.try(:position) }
      resolved_version = expected_version.resolve_for(stream, last_stream_version)<commit_msg>Basic #read implementation in RailsEventStoreActiveRecord::EventRepository<commit_after>require 'activerecord-import'

module RailsEventStoreActiveRecord
  class EventRepository
    POSITION_SHIFT = 1

    def initialize
      verify_correct_schema_present
      @repo_reader = EventRepositoryReader.new
    end

    def append_to_stream(events, stream, expected_version)
      add_to_stream(normalize_to_array(events), stream, expected_version, true) do |event|
        build_event_record(event).save!
        event.event_id
      end
    end

    def link_to_stream(event_ids, stream, expected_version)
      (normalize_to_array(event_ids) - Event.where(id: event_ids).pluck(:id)).each do |id|
        raise RubyEventStore::EventNotFound.new(id)
      end
      add_to_stream(normalize_to_array(event_ids), stream, expected_version, nil) do |event_id|
        event_id
      end
    end

    def delete_stream(stream)
      EventInStream.where(stream: stream.name).delete_all
    end

    def has_event?(event_id)
      @repo_reader.has_event?(event_id)
    end

    def last_stream_event(stream)
      @repo_reader.last_stream_event(stream)
    end

    def read_events_forward(stream, after_event_id, count)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_events_forward(stream, after_event_id, count)
    end

    def read_events_backward(stream, before_event_id, count)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_events_backward(stream, before_event_id, count)
    end

    def read_stream_events_forward(stream)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_stream_events_forward(stream)
    end

    def read_stream_events_backward(stream)
      raise ReservedInternalName if stream.name.eql?("all")
      @repo_reader.read_stream_events_backward(stream)
    end

    def read_all_streams_forward(after_event_id, count)
      @repo_reader.read_all_streams_forward(after_event_id, count)
    end

    def read_all_streams_backward(before_event_id, count)
      @repo_reader.read_all_streams_backward(before_event_id, count)
    end

    def read_event(event_id)
      @repo_reader.read_event(event_id)
    end

    def read(specification)
      case specification.direction
      when :forward
        read_stream_events_forward(RubyEventStore::Stream.new(specification.stream_name))
      when :backward
        read_stream_events_backward(RubyEventStore::Stream.new(specification.stream_name))
      end.each
    end

    private

    def add_to_stream(collection, stream, expected_version, include_global, &to_event_id)
      last_stream_version = ->(stream_) { EventInStream.where(stream: stream_.name).order("position DESC").first.try(:position) }
      resolved_version = expected_version.resolve_for(stream, last_stream_version)
{'fraction_non_alphanumeric': 0.07988110718929965, 'fraction_numerical': 0.000371540033438603, 'mean_word_length': 4.069679849340866, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3025161', 'n_tokens_mistral': 1721, 'n_tokens_neox': 1675, 'n_words': 285}