ファイル
btrc-hub/backend/app/services/counts_population.rb
T
2026-11-07 15:50:00 +09:00

338 行
13 KiB
Ruby

# Population first, then filters. Previous versions are joined before period filtering.
class CountsPopulation
attr_reader :definition, :params, :at, :history, :scope, :timezone
def initialize(params, authorisation_flg: true)
@authorisation_flg = authorisation_flg
@params = params
@definition = CountsResourceDefinition.new(params.fetch('resource', 'posts'))
@history = params.fetch('definition', 'latest') == 'history'
raise ArgumentError unless ['latest', 'history'].include?(params.fetch('definition', 'latest'))
@timezone = TZInfo::Timezone.get(params.fetch('timezone', 'UTC'))
@at = timestamp(params['execution_at']) || Time.current.utc
raise ArgumentError if at > Time.current + 1.second
@scope = definition.model.all
make_population
@scope = HistoryAccess.visible(@scope, 'v') if @authorisation_flg
apply_filters
rescue TZInfo::InvalidTimezoneIdentifier
raise ArgumentError
end
def timestamp(value, end_date: false)
return nil if value.nil? || value == ''
raise ArgumentError unless value.is_a?(String)
if value.match?(/\A\d{4}-\d{2}-\d{2}\z/)
date = Date.iso8601(value)
date += 1 if end_date
timezone.local_to_utc(Time.utc(date.year, date.month, date.day))
else
raise ArgumentError unless value.match?(/(?:Z|[+-]\d{2}:\d{2})\z/)
time = Time.iso8601(value).utc
raise ArgumentError unless time.year.between?(1900, 2100)
time
end
rescue TZInfo::AmbiguousTime, TZInfo::PeriodNotFound, Date::Error
raise ArgumentError
end
def latest_at_sql(row, cutoff)
keys = definition.keys.map { "newer.#{ _1 } = #{ row }.#{ _1 }" }.join(' AND ')
bound = definition.model.connection.quote(cutoff)
<<~SQL.squish
NOT EXISTS (SELECT 1 FROM #{ definition.table } newer
WHERE #{ keys } AND newer.version_no > #{ row }.version_no
AND newer.#{ definition.time_column } <= #{ bound })
SQL
end
def current_latest_sql(row = 'v')
keys = definition.keys.map { "newer.#{ _1 } = #{ row }.#{ _1 }" }.join(' AND ')
<<~SQL.squish
NOT EXISTS (SELECT 1 FROM #{ definition.table } newer
WHERE #{ keys } AND newer.version_no > #{ row }.version_no)
SQL
end
def changed_sql(key)
column = definition.column(key)
"(previous.id IS NOT NULL AND NOT (v.#{ column } <=> previous.#{ column }))"
end
# Aggregate sources are unrestricted; CountsAxes masks their output coordinates.
# Never serialise raw records from an unrestricted population.
def access_sql(model, row)
@authorisation_flg ? HistoryAccess.sql(model, row) : 'TRUE'
end
def axis_condition(scope, entry)
raise CountsAxes::PrivateCondition if entry['value'] == CountsAxes::MASKED_VALUE
key = entry.fetch('field')
raise ArgumentError unless definition.axes(history).include?(key)
value = entry['value']
unless value.nil?
valid = case definition.fields[key]
when :integer
value.is_a?(Integer) || (value.is_a?(String) && value.match?(/\A-?[0-9]+\z/))
when :boolean then [true, false].include?(value)
else value.is_a?(String)
end
raise ArgumentError unless valid
end
if entry.key?('grain')
unless definition.fields[key] == :datetime && CountsAxes::GRAINS.include?(entry['grain'])
raise ArgumentError
end
end
if key == 'tags'
scope = scope.where(snapshot_membership(entry.fetch('value')))
elsif key == 'changed_field'
raise ArgumentError unless history && definition.changes.include?(entry['value'])
scope = scope.where(Arel.sql(changed_sql(entry['value'])))
elsif definition.fields[key] == :datetime && entry['grain']
scope = scope.where(definition.expression(key) => nil) if entry['value'].nil?
if entry['value']
expression = CountsAxes.time_expression(definition.expression(key),
entry['grain'], timezone)
scope = scope.where("#{ expression } = ?", entry['value'])
end
else
raise ArgumentError unless definition.axes(history).include?(key)
scope = scope.where(definition.expression(key) => entry['value'])
end
scope
end
private
def make_population
table = definition.table
@scope = @scope.from("#{ table } v").select('v.*')
previous = ['rows', 'cols'].any? do |key|
QueryState.axes(params.fetch(key, '')).any? { _1['field'] == 'changed_field' }
end || (QueryState.records(params, 'drill') + QueryState.records(params, 'conditions'))
.any? { _1['field'] == 'changed_field' }
if previous
keys = definition.keys.map { "previous.#{ _1 } = v.#{ _1 }" }.join(' AND ')
@scope = @scope.joins("LEFT JOIN #{ table } previous ON #{ keys } " +
'AND previous.version_no = v.version_no - 1 AND ' +
access_sql(definition.model, 'previous'))
end
@scope = @scope.where("v.#{ definition.time_column } <= ?", at)
unless history
point = timestamp(params['at']) || at
raise ArgumentError if point > at
@scope = @scope.where("v.#{ definition.time_column } <= ?", point)
.where(Arel.sql(latest_at_sql('v', point))).where.not('v.event_type' => 'discard')
return
end
first = timestamp(params['from'])
last = timestamp(params['through'], end_date: true)
raise ArgumentError if first && last && first > last
@scope = @scope.where('v.created_at >= ?', first) if first
if last
operator = params['through'].length == 10 ? '<' : '<='
@scope = @scope.where("v.created_at #{ operator } ?", last)
end
end
def apply_filters
['resource_id', 'user'].each do |key|
value = params[key == 'user' ? 'user_id' : key]
next if value.nil? || value == ''
raise ArgumentError unless value.match?(/\A[0-9]+\z/)
@scope = @scope.where("#{ definition.expression(key) } = ?", value.to_i)
end
apply_tags if params['tags'].present? || params['tag'].present?
conditions = QueryState.records(params, 'conditions')
drills = QueryState.records(params, 'drill')
raise ArgumentError unless drills.length <= 100
drills.each do |entry|
raise ArgumentError unless (entry.keys - ['field', 'grain', 'value']).empty?
condition = entry.merge('operator' => 'axis_eq')
conditions << condition unless conditions.include?(condition)
end
raise ArgumentError unless conditions.is_a?(Array) && conditions.length <= 100
values = QueryState.records(params, 'condition_values')
values.each do |entry|
index = entry.fetch('condition')
raise ArgumentError unless index.between?(0, conditions.length - 1)
raise ArgumentError unless conditions[index]['values'] == true
end
evaluated = []
conditions.each_with_index do |entry, index|
if entry['operator'] == 'axis_eq'
next if evaluated.include?(entry)
evaluated << entry
end
if entry.key?('values')
raise ArgumentError unless entry['values'] == true
entry = entry.merge('values' => values.select { _1['condition'] == index }
.map { _1.fetch('value') })
end
apply_condition(entry)
end
end
def apply_condition(entry)
@condition ||= CountsCondition.new(self)
unless summary_private_condition?(entry)
@scope = @condition.apply(@scope, entry)
return
end
return if masked_noop?(entry)
# Evaluate private predicates only on authorised versions. On other versions,
# the entire field is the same opaque marker, regardless of its stored value.
original = @scope
visible = private_visibility_sql(entry['field'])
authorised = authorised_private_condition(original.where(Arel.sql(visible)), entry)
@scope = if masked_match?(entry)
authorised.or(original.where(Arel.sql("NOT (#{ visible })")))
else
authorised
end
end
def authorised_private_condition(scope, entry)
marker = '********'
if entry['operator'] == 'axis_eq'
return scope.none if entry['value'] == marker &&
definition.fields[entry['field']] != :string
return @condition.apply(scope, entry)
end
field = definition.filter_metadata.find { _1[:key] == entry['field'] }
return @condition.apply(scope, entry) unless field
return @condition.apply(scope, entry) if field[:type] == :string &&
field[:set_type] != 'tag_reference'
if entry['value'] == marker
return scope.none if ['eq', 'gt', 'ge', 'lt', 'le', 'range'].include?(entry['operator'])
return scope if entry['operator'] == 'ne'
end
if entry['values'].is_a?(Array) && entry['values'].include?(marker)
return scope.none if entry['operator'] == 'all'
return @condition.apply(scope, entry.merge('values' => entry['values'] - [marker]))
end
@condition.apply(scope, entry)
end
def summary_private_condition?(entry)
params.fetch('mode', 'summary') == 'summary' && !@authorisation_flg &&
!HistoryAccess.admin? && !CountsAxes::PUBLIC_FIELDS.include?(entry['field']) &&
entry['field'] != 'user.id'
end
def private_visibility_sql(field)
visible = HistoryAccess.sql(definition.model, 'v')
return visible unless field == 'changed_field'
"(#{ visible }) AND (#{ HistoryAccess.sql(definition.model, 'previous') })"
end
def masked_noop?(entry)
return false if entry['operator'] == 'axis_eq' || entry['blank']
set = ['in', 'not_in', 'any', 'all', 'none'].include?(entry['operator'])
return !entry.key?('values') if set
[entry['value'], entry['end']].all? { _1.nil? || _1 == '' }
end
def masked_match?(entry)
marker = '********'
return false if entry['blank']
value = entry['value']
case entry['operator']
when 'axis_eq', 'eq' then value == marker
when 'ne' then value != marker
when 'like'
pattern = Regexp.escape(value.to_s).gsub('%', '.*').gsub('_', '.')
/\A#{ pattern }\z/i.match?(marker)
when 'in', 'any' then entry.fetch('values', []).include?(marker)
when 'not_in', 'none' then !entry.fetch('values', []).include?(marker)
when 'all' then entry.fetch('values', []).all? { _1 == marker }
else false
end
end
def snapshot_membership(value)
kind, identifier = value.to_s.split(':', 2)
raise ArgumentError unless ['internal', 'external'].include?(kind)
raise ArgumentError unless identifier&.match?(/\A[1-9]\d*\z/)
key = kind == 'internal' ? 'tag_id' : 'external_tag_id'
definition.model.sanitize_sql_array([
"JSON_CONTAINS(v.tags_json, JSON_OBJECT('#{ key }', CAST(? AS UNSIGNED)))",
identifier.to_i])
end
def apply_tags
raise ArgumentError unless definition.resource == 'posts'
private = params.fetch('mode', 'summary') == 'summary' && !@authorisation_flg &&
!HistoryAccess.admin?
original = @scope if private
@scope = HistoryAccess.visible(@scope, 'v') if private
if params['tag'].present?
@scope = @scope.where(snapshot_membership(
"#{ params.fetch('kind', 'internal') }:#{ params['tag'] }"))
end
return unless params['tags'].present?
raw = params['tags']
tokens = raw.split.reject { _1.downcase == 'type:or' }
locale = Locale.find_by(code: params.fetch('locale', 'ja'))
raise ArgumentError unless locale
literals = PostTagSearch.literals(locale, tokens.join(' '))
identities = PostTagSearch.references_for(locale, literals)
predicates = literals.map do |literal|
# Resolve current names/aliases exactly as normal post search does. Only
# membership comes from history; snapshot names never identify a tag.
internal, external = identities.fetch(literal[:name])
parts = { 'internal' => internal, 'external' => external }.flat_map do |kind, ids|
Array(ids).map { |id| snapshot_membership("#{ kind }:#{ id }") }
end
expression = parts.empty? ? 'FALSE' : "(#{ parts.join(' OR ') })"
literal[:negative] ? "NOT #{ expression }" : expression
end
if predicates.empty?
@scope = original if private && params['tag'].blank?
return
end
any = params['match'] == 'any' || raw.split.any? { _1.downcase == 'type:or' }
@scope = @scope.where(Arel.sql("(#{ predicates.join(any ? ' OR ' : ' AND ') })"))
return unless private && params['tag'].blank?
# A masked tag collection cannot match a real tag. Negative terms match
# the opaque marker independently of any hidden snapshot membership.
hidden = any ? literals.any? { _1[:negative] } : literals.all? { _1[:negative] }
if hidden
visible = HistoryAccess.sql(definition.model, 'v')
@scope = @scope.or(original.where(Arel.sql("NOT (#{ visible })")))
end
end
end