<commit_before># TODO: how to replay a broken in the middle request
# in case of uuid subkey with a passed time value?
class Flow::Cassandra::Target < Flow::Action
  include Flow::Cassandra
  attr_reader :mapper, :catalog

  def setup!(mapper)
    if mapper.is_a? Cassandra::Mapper
      mapper.action :publisher, self
      @mapper = mapper
    else
      raise ArgumentError, "bad target: #{mapper}"
    end

    extend_name mapper.table
    build_catalog
    prepend_router
  end

  def propagate(type, data)
    if [:insert, :check].include? type
      mapper.config.before_insert.each {|it| it.call data }
    end

    log = catalog.one data

    unless log
      log = {}
      log.merge! key(data)
      log.merge! subkey(data)

      log[:data]        = mapper.one data
      log[:inserts]     = [log[:data]].compact
      log[:compactions] = 0
    end

    previous = log[:data]

    case type
    when :insert
      log[:inserts] << data
      log[:data] = data

      if previous and previous.keys.any? {|key| not data.keys.include? key }
        mapper.remove previous
      end

      converted_data = mapper.insert data
      propagate_next :insert, converted_data

      catalog.insert log
    when :remove
      remove = Cassandra::Mapper::Data::Remove.new mapper.config, data
      converted_data = remove.return!

      if log[:inserts].include? data
        log[:inserts].delete_at log[:inserts].index(data)
        log[:compactions] += 1

        if previous == data
          mapper.remove data
          new_data = log[:data] = log[:inserts].last
          mapper.insert new_data if new_data

          propagate_next :remove, converted_data
          new_data ? catalog.insert(log) : catalog.remove(log)
        else
          propagate_next :remove, converted_data
          catalog.insert log
        end

      else
        propagate_next :remove, converted_data
      end
    when :check<commit_msg>Fix cleanup of target action<commit_after># TODO: how to replay a broken in the middle request
# in case of uuid subkey with a passed time value?
class Flow::Cassandra::Target < Flow::Action
  include Flow::Cassandra
  attr_reader :mapper, :catalog

  def setup!(mapper)
    if mapper.is_a? Cassandra::Mapper
      mapper.action :publisher, self
      @mapper = mapper
    else
      raise ArgumentError, "bad target: #{mapper}"
    end

    extend_name mapper.table
    build_catalog
    prepend_router
  end

  def propagate(type, data)
    if [:insert, :check].include? type
      mapper.config.before_insert.each {|it| it.call data }
    end

    log = catalog.one data

    unless log
      log = {}
      log.merge! key(data)
      log.merge! subkey(data)

      log[:data]        = mapper.one data
      log[:inserts]     = [log[:data]].compact
      log[:compactions] = 0
    end

    previous = log[:data]

    case type
    when :insert
      log[:inserts] << data
      log[:data] = data

      if previous and previous.keys.any? {|key| not data.keys.include? key }
        mapper.remove previous
      end

      converted_data = mapper.insert data
      propagate_next :insert, converted_data

      catalog.insert log
    when :remove
      remove = Cassandra::Mapper::Data::Remove.new mapper.config, data
      converted_data = remove.return!

      if log[:inserts].include? data
        log[:inserts].delete_at log[:inserts].index(data)
        log[:compactions] += 1

        if previous == data
          mapper.remove data
          new_data = log[:inserts].last

          if new_data
            log[:data] = new_data
            mapper.insert new_data
            propagate_next :remove, converted_data
            catalog.insert log
          else
            propagate_next :remove, converted_data
            catalog.remove log
          end
        else
          propagate_next :remove, converted_data
          catalog.insert log
        end

      else
        propagate_next :remove, converted_data
      end
    when :check
{'fraction_non_alphanumeric': 0.0904295403165034, 'fraction_numerical': 0.001004772670183371, 'mean_word_length': 2.263934426229508, 'pattern_counts': {'":': 0, '<': 9, '<?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': '925678', 'n_tokens_mistral': 1197, 'n_tokens_neox': 1158, 'n_words': 324}