<commit_before>require 'active_record'

module RailsEventStoreActiveRecord
  module Legacy
    class EventRepository
      SERIALIZED_GLOBAL_STREAM_NAME = 'all'.freeze

      class LegacyEvent < ::ActiveRecord::Base
        self.primary_key = :id
        self.table_name = 'event_store_events'
      end

      private_constant :LegacyEvent

      def initialize
        warn <<-MSG
`RailsEventStoreActiveRecord::LegacyEventRepository` has been deprecated.

Please migrate to new database schema and use `RailsEventStoreActiveRecord::EventRepository`
instead:

  rails generate rails_event_store_active_record:v1_v2_migration

        MSG
      end

      def append_to_stream(events, stream, expected_version)
        validate_unsupported_expected_version(expected_version)
        validate_stream_is_empty(stream) if expected_version.none?
        validate_expected_version_number(expected_version, stream) if Integer === expected_version.version

        normalize_to_array(events).each do |event|
          data = event.to_h
          data[:stream] = stream.name
          LegacyEvent.create!(data)
        end
        self
      rescue ActiveRecord::RecordNotUnique
        raise RubyEventStore::EventDuplicatedInStream
      end

      def link_to_stream(_event_ids, _stream, _expected_version)
        raise RubyEventStore::NotSupported
      end

      def delete_stream(stream)
        LegacyEvent.where({stream: stream.name}).update_all(stream: SERIALIZED_GLOBAL_STREAM_NAME)
      end

      def has_event?(event_id)
        LegacyEvent.exists?(event_id: event_id)
      end

      def last_stream_event(stream)
        build_event_entity(LegacyEvent.where(stream: stream.name).last)
      end

      def read(spec)
        stream = LegacyEvent.order(id: order(spec))
        stream = stream.where(event_id: spec.with_ids) if spec.with_ids?
        stream = stream.where(event_type: spec.with_types) if spec.with_types?
        stream = stream.limit(spec.limit) if spec.limit?
        stream = stream.where(start_condition(spec)) unless spec.head?
        stream = stream.where(stream: spec.stream.name) unless spec.stream.global?

        if spec.batched?
          batch_reader = ->(offset, limit) { stream.offset(offset).limit(limit).map(&method(:build_event_entity)) }
          RubyEventStore::BatchEnumerator.new(spec.batch_size, spec.limit, batch_reader).each
        elsif spec.first?
          build_event_entity(stream.first)
        elsif spec.last?
          build_event_entity(stream.last)
        else
          stream.map(&method(:build_event_entity)).each
        end
      end

      def streams_of(event_id)
        LegacyEvent.where(event_id: event_id)
          .pluck(:stream)
          .map{|name| RubyEventStore::Stream.new(name)}
      end

      private

      def start_condition(specification)
        event_record =
          LegacyEvent.find_by!(event_id: specification.start)
        condition = specification.forward? ? 'id > ?' : 'id < ?'
        [condition, event_record]
      end

      def order(spec)
        spec.forward? ? 'ASC' : 'DESC'
      end


      def normalize_to_array(events)
        return events if events.is_a?(Enumerable)
        [events]<commit_msg>Count implementation in legacy AR repository<commit_after>require 'active_record'

module RailsEventStoreActiveRecord
  module Legacy
    class EventRepository
      SERIALIZED_GLOBAL_STREAM_NAME = 'all'.freeze

      class LegacyEvent < ::ActiveRecord::Base
        self.primary_key = :id
        self.table_name = 'event_store_events'
      end

      private_constant :LegacyEvent

      def initialize
        warn <<-MSG
`RailsEventStoreActiveRecord::LegacyEventRepository` has been deprecated.

Please migrate to new database schema and use `RailsEventStoreActiveRecord::EventRepository`
instead:

  rails generate rails_event_store_active_record:v1_v2_migration

        MSG
      end

      def append_to_stream(events, stream, expected_version)
        validate_unsupported_expected_version(expected_version)
        validate_stream_is_empty(stream) if expected_version.none?
        validate_expected_version_number(expected_version, stream) if Integer === expected_version.version

        normalize_to_array(events).each do |event|
          data = event.to_h
          data[:stream] = stream.name
          LegacyEvent.create!(data)
        end
        self
      rescue ActiveRecord::RecordNotUnique
        raise RubyEventStore::EventDuplicatedInStream
      end

      def link_to_stream(_event_ids, _stream, _expected_version)
        raise RubyEventStore::NotSupported
      end

      def delete_stream(stream)
        LegacyEvent.where({stream: stream.name}).update_all(stream: SERIALIZED_GLOBAL_STREAM_NAME)
      end

      def has_event?(event_id)
        LegacyEvent.exists?(event_id: event_id)
      end

      def last_stream_event(stream)
        build_event_entity(LegacyEvent.where(stream: stream.name).last)
      end

      def read(spec)
        stream = read_scope(spec)

        if spec.batched?
          batch_reader = ->(offset, limit) { stream.offset(offset).limit(limit).map(&method(:build_event_entity)) }
          RubyEventStore::BatchEnumerator.new(spec.batch_size, spec.limit, batch_reader).each
        elsif spec.first?
          build_event_entity(stream.first)
        elsif spec.last?
          build_event_entity(stream.last)
        else
          stream.map(&method(:build_event_entity)).each
        end
      end

      def count(spec)
        read_scope(spec).count
      end

      def streams_of(event_id)
        LegacyEvent.where(event_id: event_id)
          .pluck(:stream)
          .map{|name| RubyEventStore::Stream.new(name)}
      end

      private

      def read_scope(spec)
        stream = LegacyEvent.order(id: order(spec))
        stream = stream.where(event_id: spec.with_ids) if spec.with_ids?
        stream = stream.where(event_type: spec.with_types) if spec.with_types?
        stream = stream.limit(spec.limit) if spec.limit?
        stream = stream.where(start_condition(spec)) unless spec.head?
        stream = stream.where(stream: spec.stream.name) unless spec.stream.global?
        stream
      end

      def start_condition(specification)
        event_record =
          LegacyEvent.find_by!(event_id: specification.start)
        condition = specification.forward? ? 'id > ?' : 'id < ?'
        [condition, event_record]
      end

      def order(spec)
        spec.forward? ? 'ASC' : 'DESC'
      end


      def normalize_to_array(events)
        return events if events.is_a?(Enumerable)
        [events]
{'fraction_non_alphanumeric': 0.08439201451905626, 'fraction_numerical': 0.0006049606775559589, 'mean_word_length': 2.8537296037296036, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, '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': '5665406', 'n_tokens_mistral': 1937, 'n_tokens_neox': 1882, 'n_words': 364}