# Typed predicates over the same recorded fields used by Counts output.
# Multi-valued comparisons use EXISTS; inequality means no matching element.
class CountsCondition
  CHILD_TYPES = { 'id' => :integer, 'versionNo' => :integer, 'tagId' => :integer,
                  'tagGroupId' => :integer, 'autoGenerated' => :boolean,
                  'category' => :enum, 'kind' => :enum, 'sections' => :interval }.freeze

  def self.fields(definition)
    result = definition.metadata.map { _1.merge(labels: [definition.field_label(_1[:key])]) }
    visit = lambda do |fields, prefix, labels|
      fields.each do |field|
        key = [prefix, field[:key]].compact.join('.')
        path_labels = labels + [field[:label]]
        type = if prefix
                 CHILD_TYPES.fetch(field[:key], :string)
               elsif key == 'is_latest'
                 :boolean
               elsif key == 'parents'
                 :integer
               elsif key == 'deprecated'
                 :datetime
               elsif key == 'thumbnail'
                 :string
               else
                 definition.fields.fetch(key, :string)
               end
        type = :string if field[:children] || type == :json
        type = :integer if ['parents', 'members'].include?(key)
        physical = result.find { _1[:key] == key }
        nullable = prefix.present? || (physical ? physical[:nullable] : key != 'is_latest')
        nullable = false if field[:children]
        entry = { key:, type:, labels: path_labels, nullable:,
                  multiple: prefix.present? || field[:children].present?,
                  collection: field[:children].present? || type == :interval,
                  operators: CountsResourceDefinition::OPERATORS.fetch(type) }
        result.reject! { _1[:key] == key }
        result << entry
        visit.call(field[:children] || [], key, path_labels)
      end
    end
    visit.call(definition.output_fields, nil, [])
    # The raw JSON alias has the same membership semantics as the public tag field.
    if definition.resource == 'posts'
      result.find { _1[:key] == 'tags_json' }.merge!(type: :string, multiple: true, collection: true,
          operators: CountsResourceDefinition::OPERATORS.fetch(:string))
    end
    result + [{ key: 'user.id', type: :integer, nullable: true,
                labels: ['history:actor', 'counts:columns.resourceId'],
                operators: CountsResourceDefinition::OPERATORS.fetch(:integer) },
              { key: 'user.name', type: :string, nullable: true,
                labels: ['history:actor', 'history:name'],
                operators: CountsResourceDefinition::OPERATORS.fetch(:string) }]
  end

  def initialize(population)
    @population = population
    @definition = population.definition
    @connection = @definition.model.connection
    @language = Locale.find_by!(code: population.params.fetch('locale', 'ja')).language_code
  end

  def apply(scope, entry)
    raise ArgumentError unless entry.is_a?(Hash)
    raise ArgumentError if (entry.keys - ['field', 'operator', 'value', 'end', 'blank']).any?

    field = @definition.filter_metadata.find { _1[:key] == entry.fetch('field') }
    raise ArgumentError unless field && field[:operators].include?(entry.fetch('operator'))
    raise ArgumentError unless [true, false, nil].include?(entry['blank'])
    values = [entry['value'], entry['end']]
    raise ArgumentError unless values.all? { _1.nil? || (_1.is_a?(String) && _1.bytesize <= 8192) }
    if entry['operator'] != 'range' && !values[1].nil? && values[1] != ''
      raise ArgumentError
    end

    source, expression = source_for(field[:key])
    type = field[:type]
    blank = blank_sql(expression, type)
    if entry['blank']
      sql = if source && field[:collection]
              "NOT EXISTS (SELECT 1 FROM #{ source })"
            elsif source
              "NOT EXISTS (SELECT 1 FROM #{ source } WHERE NOT (#{ blank }))"
            else blank
            end
      return scope.where(Arel.sql(sql))
    end
    return scope if values.all? { _1.nil? || _1 == '' }

    parsed = values.map { parse(_1, type) }
    validate_enum!(field[:key], parsed.first) if type == :enum
    operator = entry['operator']
    if type == :interval
      return interval(scope, source, expression, parsed)
    end
    clauses = []
    binds = []
    if operator == 'range'
      ['>=', '<='].each_with_index do |comparison, index|
        next if parsed[index].nil?

        clauses << "#{ expression } #{ comparison } ?"
        binds << parsed[index]
      end
      raise ArgumentError if parsed.all? && parsed.first > parsed.last
    elsif !parsed.first.nil?
      comparison = { 'eq' => '=', 'ne' => '=', 'gt' => '>', 'lt' => '<',
                     'ge' => '>=', 'le' => '<=', 'like' => 'LIKE' }.fetch(operator)
      clauses << "#{ expression } #{ comparison } ?"
      binds << parsed.first
    end
    return scope if clauses.empty?

    predicate = clauses.join(' AND ')
    if source
      predicate = "EXISTS (SELECT 1 FROM #{ source } WHERE #{ predicate })"
      predicate = "NOT (#{ predicate })" if operator == 'ne'
    elsif operator == 'ne'
      predicate = "(NOT (#{ predicate }) OR #{ blank })"
    end
    scope.where(predicate, *binds)
  end

  private

  def parse(value, type)
    return nil if value.nil? || value == ''

    case type
    when :integer, :interval
      raise ArgumentError unless value.match?(/\A-?[0-9]{1,19}\z/)

      number = Integer(value, 10)
      raise ArgumentError unless number.between?(-9_223_372_036_854_775_808, 9_223_372_036_854_775_807)

      number
    when :boolean
      raise ArgumentError unless ['true', 'false'].include?(value)

      value == 'true' ? 1 : 0
    when :datetime then @population.timestamp(value)
    else value
    end
  end

  def validate_enum!(key, value)
    allowed = case key.split('.').last
              when 'event_type' then VersionRecord::DEFAULT_EVENT_TYPE_MAP.values
              when 'category', 'tag_category' then TagVersion.categories.keys
              when 'kind' then ['internal', 'external']
              when 'source_kind' then Material::SOURCE_KINDS
              end
    raise ArgumentError if value && allowed && !allowed.include?(value)
  end

  def blank_sql(expression, type)
    if [:string, :enum].include?(type)
      "(#{ expression } IS NULL OR #{ expression } = '')"
    else
      "#{ expression } IS NULL"
    end
  end

  def scalar(document, key, type = :string)
    path = @connection.quote("$.#{ key }")
    value = "JSON_EXTRACT(#{ document }, #{ path })"
    text = "JSON_UNQUOTE(NULLIF(#{ value }, CAST('null' AS JSON)))"
    return "CAST(#{ text } AS SIGNED)" if type == :integer
    return "CASE #{ text } WHEN 'true' THEN 1 WHEN 'false' THEN 0 END" if type == :boolean

    text
  end

  def json_table(document, path, name)
    "JSON_TABLE(COALESCE(#{ document }, JSON_ARRAY()), #{ @connection.quote(path) } " +
        "COLUMNS (item JSON PATH '$')) #{ name }"
  end

  def names_source(document, alias_name)
    source = "JSON_TABLE(COALESCE(JSON_KEYS(#{ document }), JSON_ARRAY()), '$[*]' " +
        "COLUMNS (language_code TEXT PATH '$')) #{ alias_name }"
    path = "CONCAT('$.', JSON_QUOTE(#{ alias_name }.language_code))"
    [source, "JSON_EXTRACT(#{ document }, #{ path })", "#{ alias_name }.language_code"]
  end

  def point
    @point ||= @population.history ? @population.at :
        @population.timestamp(@population.params['at']) || @population.at
  end

  def name_expression(row)
    language_path = @connection.quote("$.#{ @language.to_json }.name")
    name = "JSON_UNQUOTE(JSON_EXTRACT(#{ row }.names, #{ language_path }))"
    "COALESCE(#{ name }, #{ row }.name)"
  end

  def parent_source
    column = @definition.resource == 'posts' ? 'parent_post_ids' : 'parent_tag_ids'
    ids = "TRIM(REGEXP_REPLACE(COALESCE(v.#{ column }, ''), '[[:space:]]+', ' '))"
    document = "CONCAT('[', REPLACE(#{ ids }, ' ', ','), ']')"
    source = json_table(document, '$[*]', 'parent_entry')
    id = 'CAST(parent_entry.item AS UNSIGNED)'
    return [source, id] if @definition.resource == 'posts'

    bound = @connection.quote(point)
    source += ' LEFT JOIN tag_versions parent_version ON parent_version.tag_id = ' + id +
        " AND parent_version.created_at <= #{ bound } AND " +
        HistoryAccess.sql(TagVersion, 'parent_version') +
        ' AND NOT EXISTS (SELECT 1 FROM tag_versions newer_parent ' +
        'WHERE newer_parent.tag_id = parent_version.tag_id ' +
        "AND newer_parent.created_at <= #{ bound } " +
        'AND newer_parent.version_no > parent_version.version_no)'
    [source, id]
  end

  def post_tag_source
    source = json_table('v.tags_json', '$[*]', 'tag_snapshot')
    source += ' LEFT JOIN tag_versions referenced_tag ON referenced_tag.tag_id = ' +
        scalar('tag_snapshot.item', 'tag_id', :integer) +
        ' AND referenced_tag.version_no = ' + scalar('tag_snapshot.item', 'version_no', :integer) +
        " AND referenced_tag.created_at <= #{ @connection.quote(point) } AND " +
        HistoryAccess.sql(TagVersion, 'referenced_tag')
    source
  end

  def language_field(source, document, parts, inherited_language = nil)
    return [source, scalar(document, 'name')] if parts.empty?
    return [source, inherited_language] if parts == ['languageCode'] && inherited_language

    if parts.first == 'aliases'
      aliases = "JSON_EXTRACT(#{ document }, '$.aliases')"
      source += ' JOIN ' + json_table(aliases, '$[*]', 'alias_entry')
      child = parts[1] || 'name'
      return [source, inherited_language] if child == 'languageCode' && inherited_language
      columns = { 'scriptCode' => 'script_code', 'name' => 'name' }
      key = columns.fetch(child) { raise ArgumentError }
      return [source, scalar('alias_entry.item', key)]
    end
    key = { 'name' => 'name', 'scriptCode' => 'script_code',
            'autoGenerated' => 'auto_generated' }.fetch(parts.first) { raise ArgumentError }
    [source, scalar(document, key, CHILD_TYPES.fetch(parts.first, :string))]
  end

  def source_for(key)
    root, *parts = key.split('.')
    return [nil, "(#{ @population.latest_sql })"] if root == 'is_latest'
    if root == 'user' && parts.any?
      expression = parts.first == 'id' ? 'v.created_by_user_id' :
          '(SELECT actor.name FROM users actor WHERE actor.id = v.created_by_user_id)'
      return [nil, expression]
    end
    return [nil, 'v.thumbnail_base'] if root == 'thumbnail'
    return [nil, 'v.deprecated_at'] if root == 'deprecated'
    if root == 'parents'
      source, id = parent_source
      return [source, parts.first == 'name' ? name_expression('parent_version') : id]
    end
    if root == 'tags' || root == 'tags_json'
      return post_field(parts)
    end
    if root == 'names' || root == 'aliases'
      source, document, language = names_source('v.names', 'name_language')
      parts = ['aliases', *parts] if root == 'aliases'
      return language_field(source, document, parts, language)
    end
    if root == 'members' || root == 'placements'
      path = '$[*]' + parts.take_while { _1 == 'children' }.map { '.children[*]' }.join
      source = json_table("v.#{ root }", path, 'structure_entry')
      if root == 'members' && parts.empty?
        value = "COALESCE(#{ scalar('structure_entry.item', 'tag_id', :integer) }, " +
            "#{ scalar('structure_entry.item', 'tag_group_id', :integer) })"
        return [source, value]
      end
      leaf = parts.last || 'sectionName'
      leaf = 'sectionName' if leaf == 'children'
      column = leaf.gsub(/[A-Z]/) { "_#{ _1.downcase }" }
      return [source, scalar('structure_entry.item', column, CHILD_TYPES.fetch(leaf, :string))]
    end
    if root == 'export_paths_json'
      source, value, profile = names_source('v.export_paths_json', 'export_profile')
      return [source, parts.first == 'profile' ? profile : "JSON_UNQUOTE(#{ value })"]
    end
    return [nil, name_expression('v')] if root == 'name' && @definition.resource == 'tags'

    [nil, @definition.expression(root)]
  end

  def post_field(parts)
    source = post_tag_source
    document = 'tag_snapshot.item'
    case parts.first
    when 'names'
      names, entry, language = names_source('referenced_tag.names', 'tag_language')
      language_field("#{ source } JOIN #{ names }", entry, parts.drop(1), language)
    when 'sections'
      sections = "JSON_EXTRACT(#{ document }, '$.sections')"
      source += ' JOIN ' + json_table(sections, '$[*]', 'section_entry')
      [source, 'section_entry.item']
    when 'id'
      [source, "COALESCE(#{ scalar(document, 'tag_id', :integer) }, " +
          "#{ scalar(document, 'external_tag_id', :integer) })"]
    when 'versionNo' then [source, scalar(document, 'version_no', :integer)]
    when 'kind'
      [source, "CASE WHEN #{ scalar(document, 'tag_id') } IS NULL " +
          "THEN 'external' ELSE 'internal' END"]
    when 'category'
      [source, "COALESCE(#{ scalar(document, 'category') }, referenced_tag.category)"]
    else
      [source, "COALESCE(#{ scalar(document, 'name') }, referenced_tag.name)"]
    end
  end

  def interval(scope, source, document, bounds)
    raise ArgumentError if bounds.compact.any?(&:negative?)
    raise ArgumentError if bounds.all? && bounds.first > bounds.last

    first = scalar(document, 'begin_ms', :integer)
    last = scalar(document, 'end_ms', :integer)
    clauses = []
    values = []
    if bounds.first
      clauses << "(#{ last } IS NULL OR #{ last } > ?)"
      values << bounds.first
    end
    if bounds.last
      clauses << "#{ first } < ?"
      values << bounds.last
    end
    scope.where("EXISTS (SELECT 1 FROM #{ source } WHERE #{ clauses.join(' AND ') })", *values)
  end
end