#434

backend/app/services/counts_query.rb

@@ -14,11 +14,12 @@ class CountsQuery
 
   attr_reader :page, :limit, :mode, :axis
 
-  def initialize(scope, resource, params)
+  def initialize(scope, resource, params, latest_expression: nil)
     @table = scope.klass.table_name
     @resource = resource
     @id_column = resource == 'posts' ? 'post_id' : 'tag_id'
     @params = params
+    @latest_expression = latest_expression
     @scope = scope.except(:select, :order).joins(
         "LEFT JOIN users counts_actor ON counts_actor.id = #{ @table }.created_by_user_id")
     raise ArgumentError unless (params.keys - KEYS).empty?
@@ -43,6 +44,8 @@ class CountsQuery
   def count = @scope.count
 
   def latest_expression
+    return @latest_expression if @latest_expression
+
     <<~SQL.squish
       NOT EXISTS (SELECT 1 FROM #{ @table } newer
         WHERE newer.#{ @id_column } = #{ @table }.#{ @id_column }

backend/app/services/counts_state_query.rb

@@ -0,0 +1,293 @@
+# State at an exclusive UTC boundary, reconstructed only from recorded snapshots.
+class CountsStateQuery
+  KEYS = ['locale', 'resource', 'mode', 'at', 'from', 'before', 'grain',
+          'kind', 'tag', 'page', 'limit', 'sort'].freeze
+  MAX_PERIODS = 366
+
+  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', 'state')
+    @kind = params.fetch('kind', 'internal')
+    @grain = params.fetch('grain', 'month')
+    raise ArgumentError unless ['posts', 'tags'].include?(@resource)
+    raise ArgumentError unless ['state', 'trend', 'snapshot'].include?(@mode)
+    raise ArgumentError unless ['internal', 'external', 'all'].include?(@kind)
+    raise ArgumentError unless ['day', 'month', 'year'].include?(@grain)
+
+    @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.change(usec: 0)
+    raise ArgumentError if @at > @now
+  end
+
+  def call
+    # Keep totals, rows, and detail pages on one repeatable-read database snapshot.
+    ApplicationRecord.transaction do
+      @post_start = PostVersion.minimum(:created_at)
+      @tag_start = TagVersion.minimum(:created_at)
+      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 'snapshot'
+        result.merge(totals: totals).merge(details)
+      else
+        result.merge(totals: totals).merge(listing)
+      end
+    end
+  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 < #{ boundary } AND NOT EXISTS (
+        SELECT 1 FROM #{ table } newer
+        WHERE newer.#{ key } = #{ row }.#{ key }
+          AND newer.created_at < #{ 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 (#{ state_sql('post_versions', boundary, live: true) }),
+      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 tag_at WHERE event_type <> 'discard') 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 && first < boundary
+
+  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 = @connection.select_value("#{ prefix } SELECT COUNT(*) #{ source } #{ where }").to_i
+    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
+
+  def details
+    table = @resource == 'posts' ? 'post_versions' : 'tag_versions'
+    model = @resource == 'posts' ? PostVersion : TagVersion
+    scope = model.joins("INNER JOIN (#{ state_sql(table, quote(@at), live: true) }) " +
+                        "counts_at ON counts_at.id = #{ table }.id")
+    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
+    params = @params.slice('locale', 'sort', 'page', 'limit')
+        .merge('mode' => 'details', 'latest' => '0')
+    query = CountsQuery.new(scope, @resource, params, latest_expression: '1')
+    { count: query.count,
+      versions: CountsVersionRepr.many(query.details, resource: @resource,
+                                      language_code: @language) }
+  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