# Restores one recorded version per resource before applying record filters. class CountsStateQuery KEYS = ['locale', 'resource', 'mode', 'at', 'from', 'before', 'grain', 'kind', 'tag', 'page', 'limit', 'sort', 'axis', 'user_id', 'resource_id'].freeze MAX_PERIODS = 366 attr_reader :at def initialize(params, language_code:) raise ArgumentError unless (params.keys - KEYS).empty? raise ArgumentError unless params.values.all? { _1.is_a?(String) } @params = params @connection = ApplicationRecord.connection @language = language_code @resource = params.fetch('resource', 'posts') @mode = params.fetch('mode', 'summary') @inclusive = ['summary', 'details'].include?(@mode) @axis = params.fetch('axis', @resource == 'posts' ? 'tags' : 'month') @kind = params.fetch('kind', 'internal') @grain = params.fetch('grain', 'month') raise ArgumentError unless ['posts', 'tags'].include?(@resource) raise ArgumentError unless ['summary', 'details', 'state', 'trend', 'snapshot'].include?(@mode) raise ArgumentError unless ['internal', 'external', 'all'].include?(@kind) raise ArgumentError unless ['day', 'month', 'year'].include?(@grain) raise ArgumentError unless ['tags', 'month', 'user'].include?(@axis) if @mode == 'summary' && @axis == 'tags' && @resource != 'posts' raise ArgumentError end @page = integer('page', 1) @limit = integer('limit', 20) @tag = integer('tag') if params['tag'].present? raise ArgumentError if @limit > 100 || @page > 1_000_000 raise ArgumentError if @tag && @kind == 'all' raise ArgumentError if @mode == 'trend' && !@tag @now = Time.current.utc @at = timestamp('at') || @now raise ArgumentError if @at > @now end def call(metadata: true) @metadata = metadata # Keep totals, rows, and detail pages on one repeatable-read database snapshot. ApplicationRecord.transaction do unless @history_loaded @post_start = PostVersion.minimum(:created_at) @tag_start = TagVersion.minimum(:created_at) @history_loaded = true end result = { page: @page, limit: @limit, at: @at.iso8601(6), coverage: { basis: 'recorded_snapshots', exact: false, post_history_from: @post_start&.iso8601(6), tag_history_from: @tag_start&.iso8601(6) } } case @mode when 'trend' result.merge(trend) when 'summary' result.merge(metadata ? { totals: totals } : {}).merge(summary) when 'details', 'snapshot' result.merge(metadata ? { totals: totals } : {}).merge(details) else result.merge(metadata ? { totals: totals } : {}).merge(listing) end end end def each_page return enum_for(:each_page) unless block_given? @page = 1 first = call yield first total = ['summary', 'state', 'trend'].include?(@mode) ? first[:group_count] : first[:count] while @page * @limit < total @page += 1 yield call(metadata: false) end end def validate_query! filtered_scope(@resource) case @mode when 'summary' if @axis == 'tags' order_sql({ 'name' => 'name', 'count' => 'post_count', 'tag_id' => 'tag_id', 'kind' => 'kind', 'deprecated' => 'deprecated_at' }, 'count:desc', []) else record_query('summary') end when 'details', 'snapshot' record_query('details') end self end private def quote(value) = @connection.quote(value) def integer(key, default = nil) value = @params[key] return default if value.blank? && default raise ArgumentError unless value&.match?(/\A[1-9][0-9]{0,9}\z/) value.to_i end def timestamp(key) value = @params[key].presence return nil unless value pattern = /\A\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(?::\d{2}(?:\.\d{1,6})?)?Z\z/ raise ArgumentError unless value.match?(pattern) Date.iso8601(value.first(10)) Time.iso8601(value.length == 17 ? value.sub(/Z\z/, ':00Z') : value).utc end # Version numbers order equal-timestamp writes. Do not filter tags before this selection. def latest_condition(table, row, boundary) key = table == 'post_versions' ? 'post_id' : 'tag_id' <<~SQL.squish #{ row }.created_at #{ @inclusive ? '<=' : '<' } #{ boundary } AND NOT EXISTS ( SELECT 1 FROM #{ table } newer WHERE newer.#{ key } = #{ row }.#{ key } AND newer.created_at #{ @inclusive ? '<=' : '<' } #{ boundary } AND newer.version_no > #{ row }.version_no) SQL end def state_sql(table, boundary, live: false) sql = "SELECT v.* FROM #{ table } v WHERE #{ latest_condition(table, 'v', boundary) }" live ? "#{ sql } AND v.event_type <> 'discard'" : sql end def membership(row) key = @kind == 'external' ? 'external_tag_id' : 'tag_id' "JSON_CONTAINS(#{ row }.tags_json, JSON_OBJECT('#{ key }', #{ @tag }))" end # Deprecation keeps a tag in existence. Only an explicit discard removes it. def tag_visible(boundary) return '1 = 1' if @kind == 'external' <<~SQL.squish NOT EXISTS (SELECT 1 FROM tag_versions t WHERE t.tag_id = #{ @tag } AND t.event_type = 'discard' AND #{ latest_condition('tag_versions', 't', boundary) }) SQL end def ctes boundary = quote(@at) <<~SQL WITH post_at AS (#{ filtered_scope('posts').to_sql }), tag_at AS (#{ state_sql('tag_versions', boundary) }), nico_at AS (#{ state_sql('nico_tag_versions', boundary) }), pairs AS ( SELECT DISTINCT p.post_id, CASE WHEN entry.tag_id IS NULL THEN 'external' ELSE 'internal' END AS kind, COALESCE(entry.tag_id, entry.external_tag_id) AS tag_id FROM post_at p JOIN JSON_TABLE(CAST(p.tags_json AS JSON), '$[*]' COLUMNS ( tag_id BIGINT PATH '$.tag_id' NULL ON EMPTY ERROR ON ERROR, external_tag_id BIGINT PATH '$.external_tag_id' NULL ON EMPTY ERROR ON ERROR )) entry LEFT JOIN tag_at t ON t.tag_id = entry.tag_id WHERE COALESCE(entry.tag_id, entry.external_tag_id) IS NOT NULL AND (t.event_type IS NULL OR t.event_type <> 'discard') ) SQL end def totals row = @connection.select_one(<<~SQL) #{ ctes } SELECT (SELECT COUNT(*) FROM post_at) AS posts, (SELECT COUNT(*) FROM (#{ filtered_scope('tags').to_sql }) counted_tags) AS tags, (SELECT COUNT(*) FROM tag_at WHERE event_type <> 'discard' AND deprecated_at IS NOT NULL) AS deprecated_tags, (SELECT COUNT(*) FROM pairs) AS assignments, (SELECT COUNT(*) FROM pairs WHERE kind = 'external') AS external_assignments SQL row.transform_values!(&:to_i) row.merge('posts' => available?(@post_start, @at) ? row['posts'] : nil, 'assignments' => available?(@post_start, @at) ? row['assignments'] : nil, 'external_assignments' => available?(@post_start, @at) ? row['external_assignments'] : nil, 'tags' => available?(@tag_start, @at) ? row['tags'] : nil, 'deprecated_tags' => available?(@tag_start, @at) ? row['deprecated_tags'] : nil) end def available?(first, boundary) first && (@inclusive ? first <= boundary : first < boundary) end def listing path = quote("$.#{ JSON.generate(@language) }.name") catalog = <<~SQL , tag_counts AS ( SELECT membership.kind, membership.tag_id, COUNT(*) AS post_count FROM pairs membership GROUP BY membership.kind, membership.tag_id ), catalog AS ( SELECT 'internal' AS kind, tag_id, COALESCE(JSON_UNQUOTE(JSON_EXTRACT(names, #{ path })), name) AS name, deprecated_at FROM tag_at WHERE event_type <> 'discard' UNION ALL SELECT tc.kind, tc.tag_id, external_version.name, NULL AS deprecated_at FROM tag_counts tc LEFT JOIN tag_at internal_version ON tc.kind = 'internal' AND internal_version.tag_id = tc.tag_id LEFT JOIN nico_at external_version ON tc.kind = 'external' AND external_version.tag_id = tc.tag_id WHERE tc.kind = 'external' OR internal_version.tag_id IS NULL ) SQL filters = [] filters << "c.kind = #{ quote(@kind) }" unless @kind == 'all' filters << "c.tag_id = #{ @tag }" if @tag where = filters.any? ? "WHERE #{ filters.join(' AND ') }" : '' source = 'FROM catalog c LEFT JOIN tag_counts n ON n.kind = c.kind AND n.tag_id = c.tag_id' prefix = "#{ ctes } #{ catalog }" total = if @metadata @connection.select_value("#{ prefix } SELECT COUNT(*) #{ source } #{ where }").to_i end order = order_sql({ 'name' => 'c.name', 'count' => 'COALESCE(n.post_count, 0)', 'tag_id' => 'c.tag_id', 'kind' => 'c.kind', 'deprecated' => 'c.deprecated_at' }, 'count:desc', ['c.kind ASC', 'c.tag_id ASC']) rows = @connection.select_all(<<~SQL) #{ prefix } SELECT c.kind, c.tag_id, c.name, c.deprecated_at, COALESCE(n.post_count, 0) AS post_count #{ source } #{ where } ORDER BY #{ order } LIMIT #{ @limit } OFFSET #{ (@page - 1) * @limit } SQL { count: total, group_count: total, state_rows: rows.map do |row| { kind: row['kind'], tag_id: row['tag_id'], name: row['name'], deprecated: row['deprecated_at']&.iso8601, count: available?(@post_start, @at) ? row['post_count'].to_i : nil, at: @at.iso8601(6) } end } end # All record filters apply after selecting the latest recorded version at the point. def restored_scope(resource) model = resource == 'posts' ? PostVersion : TagVersion table = model.table_name scope = model.where(Arel.sql(latest_condition(table, table, quote(@at)))) .where.not(event_type: 'discard') if @tag scope = if resource == 'posts' scope.where("#{ membership(table) } AND #{ tag_visible(quote(@at)) }") elsif @kind == 'internal' scope.where(tag_id: @tag) else scope.none end end scope end def filtered_scope(resource) params = @params.slice('from', 'before', 'user_id', 'resource_id') .merge('mode' => 'details', 'latest' => '0', 'sort' => '') CountsQuery.new(restored_scope(resource), resource, params, latest_expression: '1').relation end def record_query(mode) params = @params.slice('locale', 'sort', 'page', 'limit') .merge('mode' => mode, 'latest' => '0', 'axis' => @axis == 'tags' ? 'month' : @axis, 'page' => @page.to_s, 'limit' => @limit.to_s) CountsQuery.new(filtered_scope(@resource), @resource, params, latest_expression: '1') end def summary return listing if @axis == 'tags' query = record_query('summary') { count: @metadata ? query.count : nil }.merge(query.summary(include_count: @metadata)) end def details query = record_query('details') { count: @metadata ? query.count : nil, versions: CountsVersionRepr.many(query.details, resource: @resource, language_code: @language, at: @at, inclusive: @inclusive) } end def next_boundary(time) case @grain when 'day' then time.beginning_of_day + 1.day when 'month' then time.beginning_of_month.next_month else time.beginning_of_year.next_year end end def periods @before = timestamp('before') || @now.beginning_of_month @from = timestamp('from') || @before.prev_year raise ArgumentError if @from >= @before || @before > @now rows = [] boundary = next_boundary(@from) while boundary <= @before raise ArgumentError if rows.length >= MAX_PERIODS format = { 'day' => '%Y-%m-%d', 'month' => '%Y-%m', 'year' => '%Y' }.fetch(@grain) rows << [boundary, (boundary - 1.second).strftime(format)] boundary = next_boundary(boundary) end rows end def trend points = periods metadata = { count: points.length, group_count: points.length, state_rows: [], from: @from.iso8601, before: @before.iso8601, grain: @grain } return metadata if points.empty? values = points.map do |boundary, period| "SELECT CAST(#{ quote(boundary) } AS DATETIME(6)) AS boundary, " + "#{ quote(period) } AS period" end.join(' UNION ALL ') order = order_sql({ 'period' => 'boundary', 'count' => 'post_count' }, 'period:asc', ['boundary ASC']) known = @post_start ? "periods.boundary > #{ quote(@post_start) }" : 'FALSE' rows = @connection.select_all(<<~SQL) WITH periods AS (#{ values }), series AS ( SELECT periods.boundary, periods.period, CASE WHEN #{ known } THEN COUNT(DISTINCT p.post_id) ELSE NULL END AS post_count FROM periods LEFT JOIN post_versions p ON p.event_type <> 'discard' AND #{ membership('p') } AND #{ latest_condition('post_versions', 'p', 'periods.boundary') } AND #{ tag_visible('periods.boundary') } GROUP BY periods.boundary, periods.period ) SELECT * FROM series ORDER BY #{ order } LIMIT #{ @limit } OFFSET #{ (@page - 1) * @limit } SQL metadata.merge(state_rows: rows.map do |row| { kind: @kind, tag_id: @tag, name: nil, period: row['period'], at: row['boundary'].utc.iso8601(6), count: row['post_count']&.to_i } end) end def order_sql(expressions, default, final) raw = @params.fetch('sort', default) sorts = raw.empty? ? [] : raw.split(',', -1).map { _1.split(':', -1) } raise ArgumentError if sorts.map(&:first).uniq.length != sorts.length raise ArgumentError if sorts.length > expressions.length order = sorts.flat_map do |key, direction, *extra| if extra.any? || !expressions.key?(key) || !['asc', 'desc'].include?(direction) raise ArgumentError end expression = expressions.fetch(key) ["(#{ expression }) IS NULL ASC", "#{ expression } #{ direction.upcase }"] end (order + final).join(', ') end end