From 8ea3d211dc58770b04c339f513e4bbbc44e3a5f0 Mon Sep 17 00:00:00 2001 From: miteruzo Date: Wed, 28 Oct 2026 19:28:11 +0900 Subject: [PATCH] #434 --- AGENTS.md | 15 ++ backend/AGENTS.md | 7 + .../controllers/concerns/counts_listing.rb | 2 +- .../controllers/counts_exports_controller.rb | 2 + .../controllers/counts_tables_controller.rb | 9 +- backend/app/models/history_access.rb | 45 +++-- .../representations/counts_version_repr.rb | 87 +++++++--- backend/app/services/counts_candidates.rb | 26 +-- backend/app/services/counts_condition.rb | 67 +++++-- backend/app/services/counts_defaults.rb | 20 ++- .../app/services/counts_export_download.rb | 9 +- backend/app/services/counts_export_jobs.rb | 163 ++++++++++++++++-- .../app/services/counts_export_projection.rb | 53 +++--- backend/app/services/counts_exporter.rb | 10 +- backend/app/services/counts_pivot_totals.rb | 6 +- backend/app/services/counts_population.rb | 31 ++-- backend/app/services/counts_query.rb | 9 +- .../services/counts_resource_definition.rb | 63 ++++++- backend/app/services/counts_state_query.rb | 10 +- backend/app/services/counts_table_export.rb | 23 ++- backend/app/services/counts_table_query.rb | 149 ++++++++++------ backend/app/services/post_tag_search.rb | 2 +- backend/lib/tasks/counts_exports.rake | 4 + docs/counts-exports.md | 64 ++++++- docs/systemd/btrc-counts-exports.service | 21 +++ frontend/AGENTS.md | 7 + .../src/components/common/SortableColumns.tsx | 41 ++--- .../src/components/common/SortableItem.tsx | 2 + .../src/components/common/SortableList.tsx | 55 +++--- .../counts/CountsAdditionalConditions.tsx | 32 ++-- frontend/src/components/counts/CountsAxes.tsx | 10 +- .../counts/CountsCandidatePicker.tsx | 8 +- .../counts/CountsConditionValues.tsx | 12 +- .../src/components/counts/CountsControls.tsx | 3 +- .../counts/CountsExportProgress.tsx | 56 +++++- .../src/components/counts/CountsFieldTree.tsx | 9 +- .../components/counts/CountsNestedTable.tsx | 12 +- .../src/components/counts/CountsResults.tsx | 10 +- .../counts/CountsSelectableTable.tsx | 52 ++++-- .../counts/CountsSummaryResults.tsx | 7 +- .../components/counts/CountsTableLayout.tsx | 6 +- .../src/components/counts/countsTable.css | 5 + frontend/src/lib/counts.ts | 44 ++++- frontend/src/lib/countsDefinitions.ts | 9 +- frontend/src/lib/countsDisplay.ts | 32 ++-- frontend/src/lib/countsTableLayout.ts | 1 + frontend/src/lib/queryKeys.ts | 3 +- frontend/src/locales/ja/counts.json | 3 + frontend/src/pages/CountsResourcePage.tsx | 7 +- 49 files changed, 967 insertions(+), 356 deletions(-) create mode 100644 docs/systemd/btrc-counts-exports.service diff --git a/AGENTS.md b/AGENTS.md index 3a18818..d5eaec8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,21 @@ 秘匿された本文・属性・識別情報・集計ラベルを公開する例外ではない。 それ以外の経路で秘匿属性を条件・軸・ソート・差分から推測できる実装は禁止する。 +## Mandatory Counts data contract + +- Counts の取得対象の ID・集合・件数・並び順を UI ロカールに依存させない。 + 言語・文字体系を抽出条件に使う場合は明示的な項目・条件として扱う。 +- API は記録済みの全言語名と別名を保持する。履歴の名前を現在の名前で置換しない。 + 言語名列は既存 Locale の言語・文字体系の組から定義し、未登録値を別言語で補完しない。 + 初期表示列だけは UI ロカールに合わせてよい。明示的な列指定・空選択を優先する。 +- 明細の内表は行集合を保持し、表示列・列順・列幅だけを表示設定として操作する。 + 内表行を抽出・除外する表示機能を追加しない。内表を参照する抽出条件は外表の + 採否だけを決定し、一致した内表行だけを残す実装は禁止する。 +- 言語名はタグ一件の言語別列とし、言語名のために内表をもう一段増やさない。 + 複数の別名は欠落させず、複数値の項目又は別名の内表として保持する。 +- JSON は対象タグの全言語登録値を保持する。CSV は選択された表示列を出力する。 + UI 翻訳済みラベルでデータをソート・識別したり、JSON の項目キーを変えたりしない。 + ## Mandatory naming contract - メソッド・関数など、実際に処理を実行する callable の名前以外の自前識別子に diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 57fcca0..98de7dc 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -18,6 +18,13 @@ - 履歴読み取りは HistoryAccess の共通認可を必須とする。書込用 relation を公開しない。 export は生成時と取得時に認可し、権限変更・除票後に古いファイルを再利用しない。 +## Mandatory Counts data contract + +- Counts のロカール非依存・内表行集合保持はルート AGENTS.md の必須契約に従う。 + UI 言語を名前の WHERE・ORDER BY へ暗黙に使わない。名前検索は記録済みの + 全言語又は明示された言語・文字体系を使う。関連名にも HistoryAccess を適用する。 + JSON は全言語名を保持し、CSV の列選択を理由に元スナップショットを削らない。 + ## Mandatory naming contract - ルート `AGENTS.md` の必須命名規約を適用する。メソッド等の callable 以外の diff --git a/backend/app/controllers/concerns/counts_listing.rb b/backend/app/controllers/concerns/counts_listing.rb index e7d5d8f..3b1c03e 100644 --- a/backend/app/controllers/concerns/counts_listing.rb +++ b/backend/app/controllers/concerns/counts_listing.rb @@ -12,7 +12,7 @@ module CountsListing result.merge!(query.summary) else result[:versions] = CountsVersionRepr.many( - query.details, resource:, language_code: locale.language_code) + query.details, resource:, language_code: locale.language_code, history_flg: true) end render json: result rescue ArgumentError, TypeError diff --git a/backend/app/controllers/counts_exports_controller.rb b/backend/app/controllers/counts_exports_controller.rb index 8c48745..0150ac8 100644 --- a/backend/app/controllers/counts_exports_controller.rb +++ b/backend/app/controllers/counts_exports_controller.rb @@ -8,6 +8,8 @@ class CountsExportsController < ApplicationController job = CountsExportJobs.start(payload, session[:counts_export_owner], locale.language_code) response.headers['Cache-Control'] = 'private, no-store' render json: { job: }, status: :accepted + rescue CountsExportJobs::WorkerUnavailable + render json: { error: 'export_worker_unavailable' }, status: :service_unavailable rescue CountsExportDownload::Busy head :too_many_requests rescue ArgumentError, TypeError, KeyError, JSON::ParserError diff --git a/backend/app/controllers/counts_tables_controller.rb b/backend/app/controllers/counts_tables_controller.rb index df0c287..bc1dea4 100644 --- a/backend/app/controllers/counts_tables_controller.rb +++ b/backend/app/controllers/counts_tables_controller.rb @@ -1,12 +1,14 @@ class CountsTablesController < ApplicationController def definition - resolve_locale! + locale = resolve_locale! definition = CountsResourceDefinition.new(params[:resource]) render json: { fields: definition.metadata, output_fields: definition.output_fields, filter_fields: definition.filter_metadata, axes: { latest: definition.axes(false), history: definition.axes(true) }, - grains: CountsAxes::GRAINS, defaults: CountsDefaults.metadata(params[:resource]) } + name_columns: definition.name_columns, + grains: CountsAxes::GRAINS, + defaults: CountsDefaults.metadata(params[:resource], locale: locale.code) } rescue ArgumentError head :bad_request end @@ -17,6 +19,9 @@ class CountsTablesController < ApplicationController rescue CountsTableQuery::TooManyColumns render json: { error: 'too_many_columns', max_columns: CountsTableQuery::MAX_COLUMNS }, status: :unprocessable_entity + rescue ActiveRecord::StatementTimeout => error + Rails.logger.warn("counts_query_timeout: #{ error.class }") + render json: { error: 'counts_query_timeout' }, status: :service_unavailable rescue ArgumentError, TypeError, KeyError, JSON::ParserError head :bad_request end diff --git a/backend/app/models/history_access.rb b/backend/app/models/history_access.rb index b1497a4..0bbee84 100644 --- a/backend/app/models/history_access.rb +++ b/backend/app/models/history_access.rb @@ -38,15 +38,38 @@ class HistoryAccess < ActiveSupport::CurrentAttributes table = model.table_name keys = KEYS.fetch(table) matches = keys.map { "boundary.#{ _1 } = #{ row }.#{ _1 }" }.join(' AND ') - clauses = ["NOT EXISTS (SELECT 1 FROM #{ table } boundary WHERE #{ matches } " + - "AND boundary.event_type = 'discard' AND boundary.version_no >= #{ row }.version_no)"] + boundaries = "SELECT #{ keys.join(', ') }, MAX(version_no) AS version_no FROM #{ table } " + + "WHERE event_type = 'discard' GROUP BY #{ keys.join(', ') }" + clauses = ["NOT EXISTS (SELECT 1 FROM (#{ boundaries }) boundary WHERE #{ matches } " + + "AND boundary.version_no >= #{ row }.version_no)"] clauses += reference_sql(table, row) if references clauses.map { "(#{ _1 })" }.join(' AND ') end def self.discard_reference(table, key, expression) - "NOT EXISTS (SELECT 1 FROM #{ table } blocked WHERE blocked.event_type = 'discard' " + - "AND #{ expression.call("blocked.#{ key }") })" + "NOT EXISTS (SELECT 1 FROM (#{ discarded_keys(table, key) }) blocked " + + "WHERE #{ expression.call("blocked.#{ key }") })" + end + + def self.discarded_keys(table, key) + "SELECT #{ key } FROM #{ table } WHERE event_type = 'discard' GROUP BY #{ key }" + end + + def self.json_reference(table, key, document) + "NOT EXISTS (SELECT 1 FROM JSON_TABLE(COALESCE(#{ document }, JSON_ARRAY()), " + + "'$[*]' COLUMNS (reference_id BIGINT PATH '$' NULL ON EMPTY)) reference_entry " + + "JOIN (#{ discarded_keys(table, key) }) blocked " + + "ON reference_entry.reference_id = blocked.#{ key })" + end + + def self.list_reference(table, key, document) + # Preserve FIND_IN_SET's exact token semantics, including empty tokens. + array = "CONCAT('[', REPLACE(JSON_QUOTE(REPLACE(COALESCE(#{ document }, ''), " + + "' ', ',')), ',', '\",\"'), ']')" + "NOT EXISTS (SELECT 1 FROM JSON_TABLE(#{ array }, '$[*]' " + + "COLUMNS (reference_id VARCHAR(255) PATH '$')) reference_entry " + + "JOIN (#{ discarded_keys(table, key) }) blocked " + + "ON reference_entry.reference_id = CAST(blocked.#{ key } AS CHAR))" end def self.reference_sql(table, row) @@ -54,16 +77,10 @@ class HistoryAccess < ActiveSupport::CurrentAttributes when 'post_versions' [['tag_versions', 'tag_id', 'tag_id'], ['nico_tag_versions', 'tag_id', 'external_tag_id']].map do |target, key, field| - discard_reference(target, key, ->(id) { - "JSON_CONTAINS(#{ row }.tags_json, JSON_OBJECT('#{ field }', #{ id }))" - }) - end + [discard_reference('post_versions', 'post_id', ->(id) { - "FIND_IN_SET(#{ id }, REPLACE(#{ row }.parent_post_ids, ' ', ',')) > 0" - })] + json_reference(target, key, "JSON_EXTRACT(#{ row }.tags_json, '$[*].#{ field }')") + end + [list_reference('post_versions', 'post_id', "#{ row }.parent_post_ids")] when 'tag_versions' - [discard_reference('tag_versions', 'tag_id', ->(id) { - "FIND_IN_SET(#{ id }, REPLACE(#{ row }.parent_tag_ids, ' ', ',')) > 0" - })] + [list_reference('tag_versions', 'tag_id', "#{ row }.parent_tag_ids")] when 'tag_description_versions' [discard_reference('tag_versions', 'tag_id', ->(id) { "#{ row }.tag_id = #{ id }" })] when 'material_versions' @@ -149,7 +166,7 @@ class HistoryAccess < ActiveSupport::CurrentAttributes # This number deliberately ignores time, identity and content filters to prevent count probes. def self.population_count(model) = model.unscoped.count - # Resource-wide totals provide a fast path; Counts computes filtered totals separately. + # Explicit whole-resource statistics; Counts pages do not run this aggregate. def self.population_counts(model) total, excluded = model.unscoped.optimizer_hints('MAX_EXECUTION_TIME(10000)').pick( Arel.sql('COUNT(*)'), diff --git a/backend/app/representations/counts_version_repr.rb b/backend/app/representations/counts_version_repr.rb index dc07efa..273e9c4 100644 --- a/backend/app/representations/counts_version_repr.rb +++ b/backend/app/representations/counts_version_repr.rb @@ -1,21 +1,31 @@ # Counts uses recorded values only; history/diff endpoints retain their own contract. class CountsVersionRepr - def self.many(rows, resource:, language_code:, at: nil, inclusive: true) - new(rows.to_a, resource:, language_code:, at:, inclusive:).serialise + # language_code remains a compatibility argument; it never selects recorded values. + def self.many(rows, resource:, language_code:, at: nil, inclusive: true, history_flg: false) + new(rows.to_a, resource:, language_code:, at:, inclusive:, history_flg:).serialise end - def initialize(rows, resource:, language_code:, at:, inclusive:) + def initialize(rows, resource:, language_code:, at:, inclusive:, history_flg:) @rows = rows HistoryAccess.require_many!(rows) @at = at @inclusive = inclusive @resource = resource - @language = language_code + @history_flg = history_flg @users = User.where(id: rows.map(&:created_by_user_id).compact).pluck(:id, :name).to_h @languages = master_names(Language) @scripts = master_names(Script) + @name_columns = CountsResourceDefinition.new('tags').name_columns @tag_versions = {} - load_tag_versions if resource == 'posts' + if resource == 'posts' + load_tag_versions + requests = @rows.flat_map do |row| + Array(json(row.tags_json)).filter_map do |entry| + [entry['external_tag_id'], parent_point(row)] if entry['external_tag_id'] + end + end.uniq + @external_versions = load_related_versions(NicoTagVersion, requests) + end load_parent_versions if resource == 'tags' end @@ -54,7 +64,7 @@ class CountsVersionRepr scope = scope.where("created_at #{ @inclusive ? '<=' : '<' } ?", @at) end scope.where(["(tag_id, version_no) IN (#{ placeholders })", *batch.flatten]) - .select(:tag_id, :version_no, :name, :category, :names).each do |version| + .select(:tag_id, :version_no, :name, :category, :names, :created_at).each do |version| @tag_versions[[version.tag_id, version.version_no]] = version end end @@ -71,15 +81,35 @@ class CountsVersionRepr end end + def locale_names(values) + @name_columns.to_h do |field| + locale = field.fetch(:name_locale) + entry = values.find do |value| + value[:language_code] == locale[:language_code] && + value[:script_code].to_s.downcase == locale[:script_code].to_s.downcase + end + [field.fetch(:key).to_sym, entry && entry[:name]] + end + end + def post(row) snapshots = Array(json(row.tags_json)).map do |entry| version = @tag_versions[[entry['tag_id'], entry['version_no']]] + version = nil if version && version.created_at > parent_point(row) + external = @external_versions[[entry['external_tag_id'], parent_point(row).iso8601(6)]] + recorded_names = version ? names(version.names) : [] + aliases = recorded_names.flat_map do |value| + value[:aliases].map { _1.merge('language_code' => value[:language_code], + 'language_name' => value[:language_name]) } + end { id: entry['tag_id'] || entry['external_tag_id'], kind: entry['tag_id'] ? 'internal' : 'external', tag_id: entry['tag_id'], external_tag_id: entry['external_tag_id'], - version_no: entry['version_no'], name: entry['name'] || version&.name, + version_no: entry['version_no'], + name: entry['name'] || version&.name || external&.fetch('name'), category: entry['category'] || version&.category, - sections: entry.fetch('sections', []), names: version ? names(version.names) : [] } + sections: entry.fetch('sections', []), names: recorded_names, aliases: } + .merge(locale_names(recorded_names)) end { title: row.title, url: row.url, thumbnail_base: row.thumbnail_base, tags: snapshots, parents: row.parent_post_ids.to_s.split.map { { id: _1.to_i } }, @@ -88,48 +118,57 @@ class CountsVersionRepr original_created_before: row.original_created_before&.iso8601 } end - def parent_point(row) = @at || row.created_at + def parent_point(row) + @history_flg ? [@at, row.created_at].compact.min : @at || row.created_at + end def load_parent_versions - @parent_versions = {} requests = @rows.flat_map do |row| row.parent_tag_ids.to_s.split.map { [_1.to_i, parent_point(row)] } end.uniq + @parent_versions = load_related_versions(TagVersion, requests) + end + + def load_related_versions(model, requests) + result = {} + table = model.table_name + comparison = @inclusive ? '<=' : '<' requests.each_slice(100) do |batch| queries = batch.map do |id, point| - scope = HistoryAccess.visible(TagVersion.where(tag_id: id)) - .where('created_at <= ?', point) - .where('NOT EXISTS (SELECT 1 FROM tag_versions newer ' + - 'WHERE newer.tag_id = tag_versions.tag_id AND newer.created_at <= ? ' + - 'AND newer.version_no > tag_versions.version_no)', point) - quoted = TagVersion.connection.quote(point.iso8601(6)) - scope.select("tag_id, name, names, #{ quoted } AS reference_point").to_sql + scope = HistoryAccess.visible(model.where(tag_id: id)) + .where("created_at #{ comparison } ?", point) + .where("NOT EXISTS (SELECT 1 FROM #{ table } newer " + + "WHERE newer.tag_id = #{ table }.tag_id AND newer.created_at #{ comparison } ? " + + "AND newer.version_no > #{ table }.version_no)", point) + quoted = model.connection.quote(point.iso8601(6)) + names = model == TagVersion ? 'names' : 'JSON_OBJECT() AS names' + scope.select("tag_id, name, #{ names }, #{ quoted } AS reference_point").to_sql end next if queries.empty? - TagVersion.connection.select_all(queries.map { "(#{ _1 })" }.join(' UNION ALL ')).each do |value| - @parent_versions[[value['tag_id'], value['reference_point']]] = value + model.connection.select_all(queries.map { "(#{ _1 })" }.join(' UNION ALL ')).each do |value| + result[[value['tag_id'], value['reference_point']]] = value end end + result end def tag(row) recorded_names = names(row.names) - selected = recorded_names.find { _1[:language_code] == @language } aliases = recorded_names.flat_map do |entry| entry[:aliases].map { _1.merge('language_code' => entry[:language_code], 'language_name' => entry[:language_name]) } end - { name: selected&.fetch(:name) || row.name, names: recorded_names, + { name: row.name, names: recorded_names, category: row.category, aliases:, parents: row.parent_tag_ids.to_s.split.map do |id| value = @parent_versions[[id.to_i, parent_point(row).iso8601(6)]] snapshots = value && json(value['names']) - name = snapshots&.dig(@language, 'name') || value&.fetch('name') - { id: id.to_i, name: } + recorded = names(snapshots) + { id: id.to_i, name: value&.fetch('name'), names: recorded }.merge(locale_names(recorded)) end, - deprecated: row.deprecated_at&.iso8601 } + deprecated: row.deprecated_at&.iso8601 }.merge(locale_names(recorded_names)) end end diff --git a/backend/app/services/counts_candidates.rb b/backend/app/services/counts_candidates.rb index eebf002..5a26c2a 100644 --- a/backend/app/services/counts_candidates.rb +++ b/backend/app/services/counts_candidates.rb @@ -23,22 +23,25 @@ class CountsCandidates internal = CountsPopulation.new(options) external = CountsPopulation.new(options.merge('resource' => 'nico_tags', 'execution_at' => internal.at.iso8601(6))) - language = Locale.find_by!(code: @params.fetch('locale', 'ja')).language_code connection = TagVersion.connection - path = connection.quote("$.#{ language.to_json }.name") - name = "COALESCE(JSON_UNQUOTE(JSON_EXTRACT(v.names, #{ path })), v.name)" - sources = [[internal, 'internal', name, 'v.category'], - [external, 'external', 'v.name', 'NULL']].map do |population, kind, title, category| - population.scope.except(:select).select(Arel.sql( + sources = [[internal, 'internal', 'v.category', 'v.names'], + [external, 'external', 'NULL', 'JSON_OBJECT()']] + sources = sources.map do |population, kind, category, names| + scope = population.scope.except(:select) + if @params['query'].present? + pattern = "%#{ ApplicationRecord.sanitize_sql_like(@params['query']) }%" + scope = scope.where("(v.name LIKE ? OR CAST(v.tag_id AS CHAR) = ? OR EXISTS (" + + "SELECT 1 FROM JSON_TABLE(COALESCE(JSON_EXTRACT(#{ names }, '$**.name'), " + + "JSON_ARRAY()), '$[*]' COLUMNS (name TEXT PATH '$')) candidate_name " + + 'WHERE candidate_name.name LIKE ?))', pattern, @params['query'], pattern) + end + scope.select(Arel.sql( "CONCAT('#{ kind }:', v.tag_id) AS value, v.tag_id AS id, " + - "#{ title } AS name, #{ category } AS category, '#{ kind }' AS kind")).to_sql + "v.name AS name, #{ names } AS names, #{ category } AS category, '#{ kind }' AS kind")) + .to_sql end clauses = [] binds = [] - if @params['query'].present? - clauses << '(name LIKE ? OR CAST(id AS CHAR) = ?)' - binds += ["%#{ ApplicationRecord.sanitize_sql_like(@params['query']) }%", @params['query']] - end if @params['category'].present? clauses << 'category = ?' binds << @params['category'] @@ -49,6 +52,7 @@ class CountsCandidates #{ where } ORDER BY kind, id LIMIT #{ LIMIT + 1 } OFFSET #{ (@page - 1) * LIMIT } SQL rows = connection.select_all(ApplicationRecord.sanitize_sql_array([sql, *binds])).to_a + rows.each { _1['names'] = JSON.parse(_1['names']) if _1['names'].is_a?(String) } { rows: rows.first(LIMIT), page: @page, more: rows.length > LIMIT, execution_at: internal.at.iso8601(6) } end diff --git a/backend/app/services/counts_condition.rb b/backend/app/services/counts_condition.rb index 5d2ec5f..280d58b 100644 --- a/backend/app/services/counts_condition.rb +++ b/backend/app/services/counts_condition.rb @@ -33,12 +33,13 @@ class CountsCondition multiple: prefix.present? || field[:children].present?, collection: field[:children].present? || type == :interval, operators: CountsResourceDefinition::OPERATORS.fetch(type) } + entry[:label_values] = field[:label_values] if field[:label_values] result.reject! { _1[:key] == key } result << entry visit.call(field[:children] || [], key, path_labels) end end - visit.call(definition.output_fields, nil, []) + visit.call(definition.condition_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, @@ -75,7 +76,6 @@ class CountsCondition @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) @@ -149,6 +149,25 @@ class CountsCondition scope.where(predicate, *binds) end + def tag_name_match(value) + source = post_tag_source + bound = @population.history ? 'v.created_at' : @connection.quote(point) + source += ' LEFT JOIN nico_tag_versions external_snapshot ON external_snapshot.tag_id = ' + + scalar('tag_snapshot.item', 'external_tag_id', :integer) + + " AND external_snapshot.created_at <= #{ bound } AND " + + @population.access_sql(NicoTagVersion, 'external_snapshot') + + ' AND NOT EXISTS (SELECT 1 FROM nico_tag_versions newer_external ' + + 'WHERE newer_external.tag_id = external_snapshot.tag_id ' + + "AND newer_external.created_at <= #{ bound } " + + 'AND newer_external.version_no > external_snapshot.version_no)' + document = "JSON_MERGE_PRESERVE(JSON_ARRAY(#{ scalar('tag_snapshot.item', 'name') }, " + + "referenced_tag.name, CONCAT('nico:', external_snapshot.name)), " + + "COALESCE(JSON_EXTRACT(referenced_tag.names, '$**.name'), JSON_ARRAY()))" + source += ' JOIN ' + json_table(document, '$[*]', 'search_name') + @definition.model.sanitize_sql_array([ + "EXISTS (SELECT 1 FROM #{ source } WHERE JSON_UNQUOTE(search_name.item) = ?)", value]) + end + private def set_condition(scope, entry, field) @@ -268,10 +287,10 @@ class CountsCondition @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)" + def name_source(row, name) + document = "JSON_MERGE_PRESERVE(JSON_ARRAY(#{ row }.name), " + + "COALESCE(JSON_EXTRACT(#{ row }.names, '$.*.name'), JSON_ARRAY()))" + [json_table(document, '$[*]', name), "JSON_UNQUOTE(#{ name }.item)"] end def parent_source @@ -282,7 +301,7 @@ class CountsCondition id = 'CAST(parent_entry.item AS UNSIGNED)' return [source, id] if @definition.resource == 'posts' - bound = @connection.quote(point) + bound = @population.history ? 'v.created_at' : @connection.quote(point) source += ' LEFT JOIN tag_versions parent_version ON parent_version.tag_id = ' + id + " AND parent_version.created_at <= #{ bound } AND " + @population.access_sql(TagVersion, 'parent_version') + @@ -298,7 +317,8 @@ class CountsCondition 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 " + + " AND referenced_tag.created_at <= #{ @population.history ? 'v.created_at' : + @connection.quote(point) } AND " + @population.access_sql(TagVersion, 'referenced_tag') source end @@ -323,6 +343,9 @@ class CountsCondition def source_for(key) root, *parts = key.split('.') + if @definition.resource == 'tags' && @definition.name_column(root) + return [nil, @definition.name_expression(root)] + end return [nil, "(#{ @population.latest_sql })"] if root == 'latest' if root == 'user' && parts.any? expression = parts.first == 'id' ? 'v.created_by_user_id' : @@ -333,7 +356,17 @@ class CountsCondition 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] + if @definition.resource == 'tags' && @definition.name_column(parts.first) + return [source, @definition.name_expression(parts.first, 'parent_version')] + end + if parts.first == 'names' + names, document, language = names_source('parent_version.names', 'parent_language') + return language_field("#{ source } JOIN #{ names }", document, parts.drop(1), language) + end + return [source, id] unless parts.first == 'name' + + names, value = name_source('parent_version', 'parent_name') + return ["#{ source } JOIN #{ names }", value] end if root == 'tags' || root == 'tags_json' return post_field(parts) @@ -360,7 +393,7 @@ class CountsCondition 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' + return name_source('v', 'recorded_name') if root == 'name' && @definition.resource == 'tags' [nil, @definition.expression(root)] end @@ -368,10 +401,14 @@ class CountsCondition def post_field(parts) source = post_tag_source document = 'tag_snapshot.item' + if @definition.name_column(parts.first) + return [source, @definition.name_expression(parts.first, 'referenced_tag')] + end case parts.first - when 'names' + when 'names', 'aliases' names, entry, language = names_source('referenced_tag.names', 'tag_language') - language_field("#{ source } JOIN #{ names }", entry, parts.drop(1), language) + fields = parts.first == 'aliases' ? ['aliases', *parts.drop(1)] : parts.drop(1) + language_field("#{ source } JOIN #{ names }", entry, fields, language) when 'sections' sections = "JSON_EXTRACT(#{ document }, '$.sections')" source += ' JOIN ' + json_table(sections, '$[*]', 'section_entry') @@ -386,7 +423,11 @@ class CountsCondition when 'category' [source, "COALESCE(#{ scalar(document, 'category') }, referenced_tag.category)"] else - [source, "COALESCE(#{ scalar(document, 'name') }, referenced_tag.name)"] + names = "JSON_MERGE_PRESERVE(JSON_ARRAY(#{ scalar(document, 'name') }, " + + 'referenced_tag.name), ' + + "COALESCE(JSON_EXTRACT(referenced_tag.names, '$.*.name'), JSON_ARRAY()))" + ["#{ source } JOIN #{ json_table(names, '$[*]', 'recorded_name') }", + 'JSON_UNQUOTE(recorded_name.item)'] end end diff --git a/backend/app/services/counts_defaults.rb b/backend/app/services/counts_defaults.rb index 61e2cdd..3e87606 100644 --- a/backend/app/services/counts_defaults.rb +++ b/backend/app/services/counts_defaults.rb @@ -4,7 +4,7 @@ class CountsDefaults 'thumbnail_base', 'tags', 'parents', 'original_created_from', 'original_created_before', 'event_type', 'user', 'created_at'].freeze - def self.entries(resource) + def self.entries(resource, locale: 'ja') result = [{ definition: 'history', mode: 'summary', params: { 'rows' => 'created_at:month', 'cols' => 'event_type', @@ -13,6 +13,17 @@ class CountsDefaults 'rows' => '', 'cols' => '', 'conditions' => '', 'tags' => '' } } details = detail_params(resource) if details + name = CountsResourceDefinition.name_key(locale) + details = details.transform_values do |value| + value.split(' ').map do |token| + if resource == 'tags' && ['name', 'parents.name'].include?(token) + token == 'name' ? name : "parents.#{ name }" + elsif resource == 'posts' && token == 'tags.name' + "tags.#{ name }" + else token + end + end.join(' ') + end ['latest', 'history'].each do |definition| result << { definition:, mode: 'details', params: details } end @@ -38,7 +49,8 @@ class CountsDefaults end def self.for(params) - settings = entries(params.fetch('resource', 'posts')).find do |entry| + defaults = entries(params.fetch('resource', 'posts'), locale: params.fetch('locale', 'ja')) + settings = defaults.find do |entry| entry[:definition] == params.fetch('definition', 'latest') && entry[:mode] == params.fetch('mode', 'summary') end&.fetch(:params) || {} @@ -55,8 +67,8 @@ class CountsDefaults settings end - def self.metadata(resource) - entries(resource).map do |entry| + def self.metadata(resource, locale: 'ja') + entries(resource, locale:).map do |entry| entry.merge(params: entry[:params].map { |key, value| { key:, value: } }) end end diff --git a/backend/app/services/counts_export_download.rb b/backend/app/services/counts_export_download.rb index 34f7a29..3123441 100644 --- a/backend/app/services/counts_export_download.rb +++ b/backend/app/services/counts_export_download.rb @@ -18,13 +18,13 @@ class CountsExportDownload end def self.verifier = Rails.application.message_verifier('counts_export_download') - def self.prepare(exporter, owner, progress: nil, check: nil, identifier: nil) + def self.prepare(exporter, owner, progress: nil, phase: nil, check: nil, identifier: nil) FileUtils.mkdir_p(directory, mode: 0o700) authority = HistoryAccess.authority slot = acquire_slot begin - generate(exporter, owner, progress:, check:, identifier:, authority:) + generate(exporter, owner, progress:, phase:, check:, identifier:, authority:) ensure slot.close end @@ -41,7 +41,8 @@ class CountsExportDownload raise Busy end - def self.generate(exporter, owner, progress: nil, check: nil, identifier: nil, authority: HistoryAccess.authority) + def self.generate(exporter, owner, progress: nil, phase: nil, check: nil, + identifier: nil, authority: HistoryAccess.authority) FileUtils.mkdir_p(directory, mode: 0o700) Dir.glob(directory.join('counts-*.export')).each do |path| File.unlink(path) if File.mtime(path) < Time.current - EXPIRY @@ -64,7 +65,7 @@ class CountsExportDownload path = file.path begin file.binmode - exporter.body(progress:).each do |chunk| + exporter.body(progress:, phase:).each do |chunk| check&.call raise TimedOut if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline diff --git a/backend/app/services/counts_export_jobs.rb b/backend/app/services/counts_export_jobs.rb index c6ee571..b3536f0 100644 --- a/backend/app/services/counts_export_jobs.rb +++ b/backend/app/services/counts_export_jobs.rb @@ -4,7 +4,46 @@ require 'fileutils' # A durable, bounded export spool. Web and supervised workers share the same POSIX volume. class CountsExportJobs class Cancelled < StandardError; end + class WorkerUnavailable < StandardError; end ACTIVE = ['queued', 'running', 'cancelling'].freeze + HEARTBEAT_SECONDS = 15 + QUEUE_SECONDS = CountsExportDownload::MAX_EXPORT_SECONDS + 30 + + def self.worker_id + @worker_id ||= ".worker-#{ Process.pid }-#{ SecureRandom.hex(8) }" + end + + def self.pulse + FileUtils.mkdir_p(directory, mode: 0o700) + write({ 'id' => worker_id, 'heartbeat_at' => Time.current.to_f }) + end + + def self.worker_available? + Dir.glob(directory.join('.worker-*.json')).any? do |path| + JSON.parse(File.read(path)).fetch('heartbeat_at') > Time.current.to_f - HEARTBEAT_SECONDS + rescue Errno::ENOENT + false + end + end + + def self.expire_stalled(state) + now = Time.current.to_f + queued = state['status'] == 'queued' + stale = if queued + now - state.fetch('created_at', now) > QUEUE_SECONDS || + (!worker_available? && now - state.fetch('created_at', now) > HEARTBEAT_SECONDS) + else + ['running', 'cancelling'].include?(state['status']) && + now - state.fetch('heartbeat_at', now) > HEARTBEAT_SECONDS + end + return unless stale + + state.merge!('status' => 'failed', 'error' => queued ? 'worker_unavailable' : 'worker_stalled', + 'updated_at' => now) + remove_download(state) + state.delete('token') + write(state) + end def self.directory = CountsExportDownload.directory.join('jobs') def self.verifier = Rails.application.message_verifier('counts_export_job') @@ -91,10 +130,14 @@ class CountsExportJobs payload[time_key] = Time.current.utc.iso8601(6) end exporter = CountsExporter.new(payload, language_code: language) + authority = HistoryAccess.authority synchronise do clean + raise WorkerUnavailable unless worker_available? active = Dir.glob(directory.join('*.json')).count do |path| - ACTIVE.include?(JSON.parse(File.read(path))['status']) + existing = JSON.parse(File.read(path)) + expire_stalled(existing) + ACTIVE.include?(existing['status']) end raise CountsExportDownload::Busy if active >= CountsExportDownload::MAX_CONCURRENT_EXPORTS @@ -102,8 +145,10 @@ class CountsExportJobs state = { 'id' => id, 'owner' => owner, 'payload' => payload, 'language' => language, 'resource' => exporter.resource, 'format' => exporter.format, 'status' => 'queued', + 'phase' => 'queued', 'created_at' => Time.current.to_f, + 'updated_at' => Time.current.to_f, 'heartbeat_at' => Time.current.to_f, 'processed' => 0, 'total' => nil, - **HistoryAccess.authority, + **authority, 'expires_at' => (Time.current + CountsExportDownload::EXPIRY).to_f } write(state) verifier.generate({ 'id' => id, 'owner' => owner }, @@ -122,12 +167,15 @@ class CountsExportJobs def self.status(token, owner, cancel: false) id = authorised_id(token, owner) return unless id + snapshot = read(id) + return unless snapshot && snapshot['owner'] == owner + return unless HistoryAccess.authorised_artifact?(snapshot) synchronise do state = read(id) return unless state && state['owner'] == owner && state['expires_at'] > Time.current.to_f - return unless HistoryAccess.authorised_artifact?(state) + expire_stalled(state) if cancel && ['queued', 'running', 'ready'].include?(state['status']) state['status'] = state['status'] == 'running' ? 'cancelling' : 'cancelled' @@ -135,24 +183,28 @@ class CountsExportJobs state.delete('token') write(state) end - state.slice('status', 'processed', 'total', 'format', 'token', 'error') + state.slice('status', 'phase', 'processed', 'total', 'format', 'token', 'error', + 'updated_at', 'heartbeat_at') end end - def self.checkpoint(id, processed = nil, total = nil) + def self.checkpoint(id, processed = nil, total = nil, phase: nil) synchronise do state = read(id) unless state && state['status'] == 'running' && state['expires_at'] > Time.current.to_f raise Cancelled end - if processed - state.merge!('processed' => processed, 'total' => total) + if processed || phase + state.merge!('processed' => processed, 'total' => total) unless processed.nil? + state['phase'] = phase if phase + state['updated_at'] = Time.current.to_f write(state) end end end def self.work_one + pulse FileUtils.mkdir_p(directory, mode: 0o700) Dir.glob(directory.join('*.json')).sort.each do |path| id = File.basename(path, '.json') @@ -171,12 +223,15 @@ class CountsExportJobs next end candidate['status'] = 'running' + candidate['phase'] = 'aggregating' + candidate['heartbeat_at'] = Time.current.to_f + candidate['updated_at'] = Time.current.to_f write(candidate) candidate end next unless state - perform(state) + supervise(state) return true end end @@ -184,6 +239,80 @@ class CountsExportJobs false end + def self.supervise(state) + id = state.fetch('id') + ApplicationRecord.connection_handler.clear_all_connections! + pid = Process.fork do + Signal.trap('TERM') { raise Cancelled } + Signal.trap('INT') { raise Cancelled } + begin + perform(state) + ensure + ApplicationRecord.connection_handler.clear_all_connections! + Process.exit!(0) + end + end + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + termination = nil + reaped = false + begin + loop do + if Process.waitpid(pid, Process::WNOHANG) + reaped = true + break + end + pulse + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + terminal = synchronise do + current = read(id) + if current && current['status'] == 'running' + if now - started > CountsExportDownload::MAX_EXPORT_SECONDS + current.merge!('status' => 'failed', 'error' => 'generation_timeout') + end + current['heartbeat_at'] = Time.current.to_f + write(current) + end + !current || !['running', 'ready'].include?(current['status']) + end + begin + if terminal && termination.nil? + Process.kill('TERM', pid) + termination = now + elsif termination && now - termination > 5 + Process.kill('KILL', pid) + end + rescue Errno::ESRCH + # The child may exit between the nonblocking wait and signal delivery. + nil + end + sleep(1) + end + ensure + unless reaped + begin + Process.kill('KILL', pid) + rescue Errno::ESRCH + nil + end + Process.waitpid(pid) + end + synchronise do + current = read(id) + if current && current['status'] != 'ready' + if ['running', 'cancelling'].include?(current['status']) + current['status'] = current['status'] == 'cancelling' ? 'cancelled' : 'failed' + current['error'] = 'interrupted' if current['status'] == 'failed' + end + remove_download(current) + remove_partial(id) + current.delete('token') + current['updated_at'] = Time.current.to_f + write(current) + end + end + end + end + def self.perform(state) actor = state['actor_id'] && User.find_by(id: state['actor_id']) actor = nil unless state['admin'] && actor&.admin? && !actor.banned? @@ -221,18 +350,21 @@ class CountsExportJobs last = now end end + phase = ->(value) { checkpoint(id, phase: value) } token = CountsExportDownload.prepare(exporter, state.fetch('owner'), - progress:, check:, identifier: id) + progress:, phase:, check:, identifier: id) + file = CountsExportDownload.find(token, state.fetch('owner')) + raise Cancelled unless file + synchronise do current = read(id) if (!current || current['status'] != 'running' || current['expires_at'] <= Time.current.to_f) - file = CountsExportDownload.find(token, state.fetch('owner')) - File.unlink(file.fetch('path')) if file + File.unlink(file.fetch('path')) if File.exist?(file.fetch('path')) raise Cancelled end - file = CountsExportDownload.find(token, state.fetch('owner')) current.merge!('status' => 'ready', 'token' => token, + 'updated_at' => Time.current.to_f, 'download_file' => File.basename(file.fetch('path'))) current['total'] = current['processed'] write(current) @@ -242,8 +374,11 @@ class CountsExportJobs synchronise do current = read(id) if current - current['status'] = error.is_a?(Cancelled) ? 'cancelled' : 'failed' - current['error'] = error.class.name + unless current['status'] == 'failed' + current['status'] = error.is_a?(Cancelled) ? 'cancelled' : 'failed' + current['error'] = error.class.name + end + current['updated_at'] = Time.current.to_f write(current) end end diff --git a/backend/app/services/counts_export_projection.rb b/backend/app/services/counts_export_projection.rb index 7ec5398..7a8dab2 100644 --- a/backend/app/services/counts_export_projection.rb +++ b/backend/app/services/counts_export_projection.rb @@ -18,10 +18,10 @@ class CountsExportProjection 'autoGenerated' => 'auto_generated' }.freeze NAME_FIELDS = ['languageCode', 'scriptCode', 'name', 'aliases', 'autoGenerated'].freeze CHILDREN = { - 'tags' => ['id', 'versionNo', 'kind', 'name', 'category', 'sections', 'names'], + 'tags' => ['id', 'versionNo', 'kind', 'name', 'category', 'sections', 'names', 'aliases'], 'names' => NAME_FIELDS, 'aliases' => ['languageCode', 'name', 'scriptCode'], - 'parents' => ['id', 'name'] }.freeze + 'parents' => ['id', 'name', 'names'] }.freeze def self.default_columns(resource) fields = { 'posts' => ['title', 'url', 'tags', 'parents', 'video_ms'], @@ -47,6 +47,14 @@ class CountsExportProjection params = CountsDefaults.for(params.merge('resource' => resource)).merge(params) end @params = params + definition = CountsResourceDefinition.new(resource) + name_keys = definition.name_columns.map { _1[:key] } + @root_keys = ROOT_KEYS.merge(name_keys.to_h { [_1, _1] }) + @child_keys = CHILD_KEYS.merge(name_keys.to_h { [_1, _1] }) + @children = CHILDREN.transform_values(&:dup) + ['tags', 'parents'].each { @children[_1] += name_keys } + @complete_names = params.fetch('format', 'json') == 'json' + @resource = resource @labels = JSON.parse(params.fetch('labels', '{}')) raise ArgumentError unless @labels.is_a?(Hash) && @labels.length <= 100 raise ArgumentError unless @labels.all? { |key, value| valid_label?(key, value) } @@ -54,7 +62,7 @@ class CountsExportProjection allowed = if summary tags ? ['kind', 'tag_id', 'name', 'deprecated', 'count'] : ['group', 'count'] else - CountsQuery::COLUMNS.fetch(resource) + CountsQuery::COLUMNS.fetch(resource) + (resource == 'tags' ? name_keys : []) end defaults = summary ? allowed : self.class.default_columns(resource) @columns = QueryState.list(params.fetch('columns', defaults.join(' '))) @@ -77,7 +85,7 @@ class CountsExportProjection end @selected = params.key?('children') ? QueryState.list(params['children']).to_set : self.class.default_children(resource).to_set - valid = CHILDREN.flat_map do |key, fields| + valid = @children.flat_map do |key, fields| paths(fields, key) end raise ArgumentError if @selected && (@selected.to_a - valid).any? @@ -91,9 +99,9 @@ class CountsExportProjection @selected.delete(path) unless children.any? { @selected.include?("#{ path }.#{ _1 }") } end end - CHILDREN.each { |key, fields| prune.call(fields, key) } + @children.each { |key, fields| prune.call(fields, key) } pairs = @columns.zip(@headers).select do |key, _header| - !CHILDREN.key?(key) || CHILDREN[key].any? { @selected.include?("#{ key }.#{ _1 }") } + !@children.key?(key) || @children[key].any? { @selected.include?("#{ key }.#{ _1 }") } end @columns = pairs.map(&:first) @headers = pairs.map(&:last) @@ -110,24 +118,28 @@ class CountsExportProjection def metadata @columns.each_with_index.map do |key, index| - { key: @summary ? key : ROOT_KEYS.fetch(key), label: @headers[index] } + { key: @summary ? key : @root_keys.fetch(key), label: @headers[index] } end end def project(row) - @columns.to_h do |key| + result = @columns.to_h do |key| if @summary value = key == 'group' ? { key: row[:key], name: row[:name] } : row[key.to_sym] [key, value] else value = case key when 'thumbnail' then row[:thumbnail_base] - else row[ROOT_KEYS.fetch(key).to_sym] + else row[@root_keys.fetch(key).to_sym] end - value = nested(value || [], CHILDREN.fetch(key), key) if CHILDREN.key?(key) - [ROOT_KEYS.fetch(key), value] + value = nested(value || [], @children.fetch(key), key) if @children.key?(key) + [@root_keys.fetch(key), value] end end + if @complete_names && ((!@summary && @resource == 'tags') || row.key?(:names)) + result['names'] = row[:names] || [] + end + result end def csv_record(record, row) @@ -165,7 +177,7 @@ class CountsExportProjection def fields_for(key) return NAME_FIELDS if key == 'names' - return ['name', 'scriptCode'] if key == 'aliases' + return ['languageCode', 'name', 'scriptCode'] if key == 'aliases' [] end @@ -197,13 +209,10 @@ class CountsExportProjection end def sort_value(row, key) - if key == 'languageCode' - language = named_code(row, 'language') - return @labels.fetch("language:#{ language[:code] }", language[:name]) - end - return named_code(row, 'script')[:name] if key == 'scriptCode' + return fetch(row, 'language_code') if key == 'languageCode' + return fetch(row, 'script_code') if key == 'scriptCode' - fetch(row, CHILD_KEYS.fetch(key)) + fetch(row, @child_keys.fetch(key)) end def comparison(left, right) @@ -249,8 +258,8 @@ class CountsExportProjection ordered = (@order + keys.map { "#{ prefix }.#{ _1 }" }).uniq .filter_map { |path| keys.find { "#{ prefix }.#{ _1 }" == path } } sorted(rows, prefix, fields).map do |row| - ordered.to_h do |key| - stable = CHILD_KEYS.fetch(key) + result = ordered.to_h do |key| + stable = @child_keys.fetch(key) value = case key when 'languageCode' then named_code(row, 'language') when 'scriptCode' then named_code(row, 'script') @@ -261,6 +270,10 @@ class CountsExportProjection end [stable, value] end + if @complete_names && ['tags', 'parents'].include?(prefix) + result['names'] = fetch(row, 'names') || [] + end + result end end end diff --git a/backend/app/services/counts_exporter.rb b/backend/app/services/counts_exporter.rb index 5752b35..09cd584 100644 --- a/backend/app/services/counts_exporter.rb +++ b/backend/app/services/counts_exporter.rb @@ -48,17 +48,19 @@ class CountsExporter end end - def body(progress: nil) - return @table_export.body(progress:) if @table_export + def body(progress: nil, phase: nil) + return @table_export.body(progress:, phase:) if @table_export Enumerator.new do |output| ApplicationRecord.transaction(isolation: :repeatable_read) do processed = 0 first_row = true started = false + phase&.call('aggregating') pages do |page| total = @mode == 'details' ? page[:count] : nil progress&.call(processed, total) + phase&.call('writing') unless started meta = { resource:, mode: @mode, basis: @history ? 'history' : 'state', at: @query_params['at'], boundary: 'inclusive', @@ -90,6 +92,7 @@ class CountsExporter processed += 1 progress&.call(processed, total) end + phase&.call('aggregating') end progress&.call(processed, processed) output << ']}' if format == 'json' @@ -135,7 +138,8 @@ class CountsExporter { count: total, versions: CountsVersionRepr.many(query.details, resource:, language_code: @language, - at: Time.iso8601(@query_params['at'])) } + at: Time.iso8601(@query_params['at']), + history_flg: true) } end total ||= result[:group_count] yield result diff --git a/backend/app/services/counts_pivot_totals.rb b/backend/app/services/counts_pivot_totals.rb index 7986d31..44c5978 100644 --- a/backend/app/services/counts_pivot_totals.rb +++ b/backend/app/services/counts_pivot_totals.rb @@ -8,6 +8,9 @@ class CountsPivotTotals end def call + # MySQL's unlimited LIMIT prevents CTE merging: authorise once for all branches. + basis = @population.scope.except(:select, :order).select('v.*') + .limit(18_446_744_073_709_551_615).to_sql queries = @rows.flat_map do |row_depth, headers| @columns.filter_map do |column_depth, columns| next if row_depth == @axes.rows.length && column_depth == @axes.columns.length @@ -15,7 +18,7 @@ class CountsPivotTotals indexes = (0...row_depth).to_a + (0...column_depth).map { @axes.rows.length + _1 } expressions = indexes.map { @axes.expressions[_1] } - scope = @population.scope.except(:select, :order) + scope = @population.scope.except(:select, :order, :where, :from).from('authorised v') joins = @axes.joins(indexes.map { @axes.all[_1].fetch('field') }) scope = scope.joins(joins) unless joins.empty? if row_depth.positive? @@ -48,6 +51,7 @@ class CountsPivotTotals sql = queries.map { "(#{ _1 })" }.join(' UNION ALL ') @population.definition.model.connection.select_all( + "WITH authorised AS (#{ basis }) " + "SELECT /*+ MAX_EXECUTION_TIME(10000) */ * FROM (#{ sql }) totals").map do |record| row_depth = record['row_depth'].to_i column_depth = record['column_depth'].to_i diff --git a/backend/app/services/counts_population.rb b/backend/app/services/counts_population.rb index c3d8fb4..97611db 100644 --- a/backend/app/services/counts_population.rb +++ b/backend/app/services/counts_population.rb @@ -67,10 +67,15 @@ class CountsPopulation def make_population table = definition.table @scope = @scope.from("#{ table } v").select('v.*') - 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')) + previous = ['rows', 'cols'].any? do |key| + QueryState.axes(params.fetch(key, '')).any? { _1['field'] == 'changed_field' } + end || QueryState.records(params, 'drill').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 @@ -169,15 +174,21 @@ class CountsPopulation end return unless params['tags'].present? - locale = Locale.find_by!(code: params.fetch('locale', 'ja')) refs = PostTagSearch.parse_references(params, params['tags'].split) raw = params['tags'] - literals = PostTagSearch.literals(locale, raw, references: refs) + literals = PostTagSearch.literals(nil, raw, references: refs) predicates = literals.map do |literal| - internal, external = PostTagSearch.references(locale, literal) - parts = Array(internal).map { snapshot_membership("internal:#{ _1 }") } + - Array(external).map { snapshot_membership("external:#{ _1 }") } - expression = parts.empty? ? 'FALSE' : "(#{ parts.join(' OR ') })" + reference = literal[:reference] + expression = if reference + parts = { 'internal' => reference['tag_id'], + 'external' => reference['external_tag_id'] }.filter_map do |kind, id| + snapshot_membership("#{ kind }:#{ id }") if id + end + parts.empty? ? 'FALSE' : "(#{ parts.join(' OR ') })" + else + @condition ||= CountsCondition.new(self) + @condition.tag_name_match(literal[:name]) + end literal[:negative] ? "NOT #{ expression }" : expression end return if predicates.empty? diff --git a/backend/app/services/counts_query.rb b/backend/app/services/counts_query.rb index 742c460..3525d2e 100644 --- a/backend/app/services/counts_query.rb +++ b/backend/app/services/counts_query.rb @@ -129,9 +129,8 @@ class CountsQuery 'tags' => json_order('post_versions.tags_json'), 'parents' => list_order('post_versions.parent_post_ids')) else - name = "JSON_UNQUOTE(JSON_EXTRACT(tag_versions.names, #{ locale_name_path }))" common.merge( - 'name' => ["COALESCE(#{ name }, tag_versions.name)"], + 'name' => ['tag_versions.name'], 'names' => json_order('tag_versions.names'), 'category' => ['tag_versions.category'], 'aliases' => json_order( @@ -141,12 +140,6 @@ class CountsQuery end end - def locale_name_path - language = Locale.find_by(code: @params.fetch('locale', 'ja'))&.language_code || 'ja' - path = "$.#{ JSON.generate(language) }.name" - @scope.klass.connection.quote(path) - end - # Multi-valued parent cells sort by cardinality, then binary snapshot text. def json_order(column) [ diff --git a/backend/app/services/counts_resource_definition.rb b/backend/app/services/counts_resource_definition.rb index cbe7f0b..274f925 100644 --- a/backend/app/services/counts_resource_definition.rb +++ b/backend/app/services/counts_resource_definition.rb @@ -111,6 +111,30 @@ class CountsResourceDefinition def field_label(key) = LABELS.fetch(key, "counts:attributes.#{ key }") + def self.name_key(code) = "name_#{ code.to_s.downcase.tr('-', '_') }" + + def name_columns + @name_columns ||= Locale.order(:code).map do |locale| + { key: self.class.name_key(locale.code), label: 'counts:localeName', + label_values: { locale: locale.name }, + name_locale: { code: locale.code, language_code: locale.language_code, + script_code: locale.script_code } } + end + end + + def name_column(key) = name_columns.find { _1[:key] == key } + + def name_expression(key, row = 'v') + locale = name_column(key).fetch(:name_locale) + connection = model.connection + path = connection.quote("$.#{ locale[:language_code].to_json }") + script = connection.quote(locale[:script_code]) + "CASE WHEN LOWER(JSON_UNQUOTE(JSON_EXTRACT(#{ row }.names, " + + "CONCAT(#{ path }, '.script_code')))) = LOWER(#{ script }) " + + "THEN JSON_UNQUOTE(NULLIF(JSON_EXTRACT(#{ row }.names, " + + "CONCAT(#{ path }, '.name')), CAST('null' AS JSON))) END" + end + def output_fields content = case resource when 'posts' @@ -122,7 +146,9 @@ class CountsResourceDefinition keys = ['resource_id', 'version_no', 'latest'] + (content - ['thumbnail']) + ['event_type', 'user', 'created_at'] keys.unshift('thumbnail') if content.include?('thumbnail') - keys.map do |key| + keys.flat_map do |key| + next name_columns if key == 'name' && resource == 'tags' + next [] if key == 'names' && resource == 'tags' label = if key == 'parents' resource == 'tags' ? 'tags:parents' : 'shared:labels.parentPosts' else field_label(key) @@ -130,16 +156,15 @@ class CountsResourceDefinition field = { key:, label: } children = child_fields(key) field[:children] = children if children.any? - field + [field] end end def child_fields(key, depth = 4) keys = case key - when 'tags' then ['id', 'versionNo', 'kind', 'name', 'category', 'sections', 'names'] + when 'tags' then ['id', 'versionNo', 'kind', 'name', 'category', 'sections', 'aliases'] when 'names' then ['languageCode', 'scriptCode', 'name', 'aliases', 'autoGenerated'] - when 'aliases' - resource == 'tags' ? ['languageCode', 'name', 'scriptCode'] : ['name', 'scriptCode'] + when 'aliases' then ['languageCode', 'name', 'scriptCode'] when 'parents' then resource == 'tags' ? ['id', 'name'] : ['id'] when 'members' then ['tagId', 'tagGroupId', 'versionNo'] when 'export_paths_json' then ['profile', 'path'] @@ -148,17 +173,39 @@ class CountsResourceDefinition scalar + (depth > 0 ? ['children'] : []) else [] end - keys.map do |child| + keys.flat_map do |child| + if child == 'name' && ['tags', 'parents'].include?(key) + next [{ key: child, label: CHILD_LABELS.fetch(child) }, *name_columns] + end label = key == 'members' && child == 'tagId' ? 'counts:columns.resourceId' : CHILD_LABELS.fetch(child) field = { key: child, label: } children = child_fields(child, child == 'children' ? depth - 1 : depth) # Language is attached to top-level aliases, not to aliases inside a language entry. - children = children.reject { _1[:key] == 'languageCode' } if child == 'aliases' + if child == 'aliases' && key == 'names' + children = children.reject { _1[:key] == 'languageCode' } + end field[:children] = children if children.any? - field + [field] end end + def condition_fields + fields = output_fields + if resource == 'tags' + fields += [{ key: 'name', label: 'counts:columns.name' }, + { key: 'names', label: 'counts:columns.names', children: child_fields('names') }] + end + if resource == 'posts' + fields = fields.map do |field| + next field unless field[:key] == 'tags' + + field.merge(children: field[:children] + [ + { key: 'names', label: 'counts:columns.names', children: child_fields('names') }]) + end + end + fields + end + def filter_metadata = @filter_metadata ||= CountsCondition.fields(self) end diff --git a/backend/app/services/counts_state_query.rb b/backend/app/services/counts_state_query.rb index 0c4f1e5..b141718 100644 --- a/backend/app/services/counts_state_query.rb +++ b/backend/app/services/counts_state_query.rb @@ -201,17 +201,16 @@ class CountsStateQuery 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, + SELECT 'internal' AS kind, tag_id, name, names, deprecated_at FROM tag_at WHERE event_type <> 'discard' UNION ALL - SELECT tc.kind, tc.tag_id, external_version.name, NULL AS deprecated_at + SELECT tc.kind, tc.tag_id, external_version.name, JSON_OBJECT() AS names, + 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 @@ -235,12 +234,13 @@ class CountsStateQuery ['c.kind ASC', 'c.tag_id ASC']) rows = @connection.select_all(<<~SQL) #{ prefix } - SELECT c.kind, c.tag_id, c.name, c.deprecated_at, + SELECT c.kind, c.tag_id, c.name, c.names, 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'], + names: row['names'].is_a?(String) ? JSON.parse(row['names']) : row['names'], deprecated: row['deprecated_at']&.iso8601, count: available?(@post_start, @at) ? row['post_count'].to_i : nil, at: @at.iso8601(6) } diff --git a/backend/app/services/counts_table_export.rb b/backend/app/services/counts_table_export.rb index 16bdbe7..4096f3f 100644 --- a/backend/app/services/counts_table_export.rb +++ b/backend/app/services/counts_table_export.rb @@ -24,7 +24,7 @@ class CountsTableExport @child_sort = QueryState.list(params.fetch('child_sort', '')).map { _1.split(':', -1) } @child_sort.each do |path, direction, *extra| raise ArgumentError if extra.any? || !['asc', 'desc'].include?(direction) - raise ArgumentError unless path&.match?(/\A[a-zA-Z_]+(?:\.[a-zA-Z]+)*\/[a-zA-Z]+\z/) + raise ArgumentError unless path&.match?(/\A[a-zA-Z_]+(?:\.[a-zA-Z_]+)*\/[a-zA-Z_]+\z/) end @format = params.fetch('format', 'json') @resource = params.fetch('resource', 'posts') @@ -55,21 +55,26 @@ class CountsTableExport end end - def body(progress: nil) + def body(progress: nil, phase: nil) Enumerator.new do |output| ApplicationRecord.transaction(isolation: :repeatable_read) do @language_names = Language.pluck(:code, :name).to_h processed = 0 started = false first = true - @query.each_page do |page| + @query.each_page(before_page: -> { phase&.call('aggregating') }) do |page| total = @summary ? nil : page[:count] progress&.call(processed, total) + phase&.call('writing') unless started - metadata = page.except(:versions, :cells, :axis_labels, :column_values, + metadata = page.except(:versions, :cells, :axis_labels, :axis_names, :column_values, :row_headers, :column_headers, :pivot_totals).merge( resource:, record_unit: @summary && format == 'json' ? 'pivot_cell' : 'row', body_row_totals: 'once_per_logical_group', page_grand_total_repeated: false, + logical_count_unit: 'regular_result_rows', + name_collection: 'all_recorded_languages', + name_columns: @query.population.definition.name_columns, + aggregate_records_in_logical_count: false, filters: @params.select { |key, _value| CountsTableQuery::KEYS.include?(key) || key == 'measures' || QueryState.valid_key?(key) }, fields: @query.population.definition.metadata, @@ -92,7 +97,7 @@ class CountsTableExport labels = pivot_row_labels(header, page) values = labels + records.map { _1[:count] } output << CSV.generate_line(values.map { safe(_1) }, row_sep: "\r\n") - processed += records.length + processed += 1 progress&.call(processed, nil) else records.each do |record| @@ -232,7 +237,15 @@ class CountsTableExport else 'data' end + values = row[:values].each_with_index.to_h { |value, index| ["a#{ index }", value] } + column[:values].each_with_index do |value, index| + values["a#{ @query.axes.rows.length + index }"] = value + end + names = page.fetch(:axis_names, {}).to_h do |key, entries| + [key, entries.fetch(values[key].to_s, {})] + end { row: row[:values], column: column[:values], kind:, + axis_names: names, row_kind: row[:kind], column_kind: column[:kind], row_depth: row[:values].length, column_depth: column[:values].length, label_key: CountsPivotLayout.label_key(@query.axes.rows.length, @query.axes.columns.length, diff --git a/backend/app/services/counts_table_query.rb b/backend/app/services/counts_table_query.rb index 6895572..258a9db 100644 --- a/backend/app/services/counts_table_query.rb +++ b/backend/app/services/counts_table_query.rb @@ -29,6 +29,7 @@ class CountsTableQuery @sort = QueryState.list(params.fetch('sort', '')).map { _1.split(':', -1) } allowed = mode == 'summary' ? axes.all.each_index.map { "a#{ _1 }" } + ['count'] : @definition.fields.keys + ['latest'] + + (@definition.resource == 'tags' ? @definition.name_columns.map { _1[:key] } : []) + (@definition.resource == 'posts' ? ['tags', 'parents'] : []) + (@definition.resource == 'tags' ? ['parents', 'deprecated'] : []) raise ArgumentError if @sort.map(&:first).uniq.length != @sort.length @@ -49,25 +50,31 @@ class CountsTableQuery result.merge(visible_output_count: visible, **totals, output_count_scope: 'before_authorisation', - output_count_unit: 'table_body_rows', + output_count_unit: 'regular_result_rows', page_grand_total_repeated: mode == 'summary' && axes.rows.any? && result[:group_count] > limit) end end def metadata - @population_counts ||= HistoryAccess.population_counts(@definition.model) - @population_counts.merge( + { population_count: nil, population_count_computed: false, excluded_count: nil, excluded_count_scope: 'current_query', population_count_scope: 'all_recorded_versions', page:, limit:, execution_at: population.at.iso8601(6), at: @params['at'].presence, definition: population.history ? 'history' : 'latest', timezone: population.timezone.identifier, rows: axes.rows, cols: axes.columns, fields: @definition.metadata, - count: population.scope.except(:select).optimizer_hints('MAX_EXECUTION_TIME(10000)').count) + count: record_counts[:visible] } end def summary + if axes.all.empty? + headers = [{ values: [], kind: 'data', depth: 0, axis_count: 0, label_key: nil }] + return { group_count: 1, visible_output_count: 1, column_values: [[]], + row_headers: page == 1 ? headers : [], column_headers: headers, pivot_totals: [], + cells: page == 1 ? [{ values: [], count: record_counts[:visible] }] : [], + axis_labels: {} } + end expressions = axes.expressions selected = expressions.each_with_index.map do |expression, index| "#{ expression } AS a#{ index }" @@ -107,10 +114,6 @@ class CountsTableQuery grouped = "SELECT #{ row_selection } FROM cells #{ row_group }" selection = (['grouped.*'] + extras).join(', ') group_sql = "SELECT #{ selection } FROM (#{ grouped }) grouped #{ measure_joins.join(' ') }" - total = @connection.select_value(<<~SQL).to_i - #{ prefix } - SELECT /*+ MAX_EXECUTION_TIME(10000) */ COUNT(*) FROM (#{ group_sql }) row_groups - SQL join = row_keys.map { "cells.#{ _1 } <=> page_rows.#{ _1 }" }.join(' AND ') join = 'TRUE' if row_keys.empty? order_cols = @sort.filter_map do |key, direction| @@ -120,22 +123,24 @@ class CountsTableQuery end order = ['page_rows.position'] + order_cols + axes.all.each_index.map { axis_order_sql("a#{ _1 }", qualifier: 'cells') } - records = @connection.select_all(<<~SQL) + result = @connection.select_all(<<~SQL).to_a #{ prefix }, row_groups AS (#{ group_sql }), numbered AS (SELECT row_groups.*, ROW_NUMBER() OVER (ORDER BY #{ order_rows.join(', ') }) AS position FROM row_groups), - page_rows AS (SELECT * FROM numbered - WHERE position > #{ (page - 1) * limit } AND position <= #{ page * limit }) - SELECT /*+ MAX_EXECUTION_TIME(10000) */ cells.* FROM cells JOIN page_rows ON #{ join } + group_totals AS (SELECT COUNT(*) AS group_count FROM row_groups) + SELECT /*+ MAX_EXECUTION_TIME(10000) */ cells.*, + page_rows.position AS counts_position, group_totals.group_count AS counts_group_count + FROM group_totals LEFT JOIN numbered page_rows + ON page_rows.position > #{ (page - 1) * limit } + AND page_rows.position <= #{ page * limit + 1 } + LEFT JOIN cells ON #{ join } ORDER BY #{ order.join(', ') } SQL - next_row = @connection.select_one(<<~SQL) - #{ prefix }, row_groups AS (#{ group_sql }), - numbered AS (SELECT row_groups.*, - ROW_NUMBER() OVER (ORDER BY #{ order_rows.join(', ') }) AS position FROM row_groups) - SELECT /*+ MAX_EXECUTION_TIME(10000) */ * FROM numbered - WHERE position = #{ page * limit + 1 } - SQL + total = result.first.fetch('counts_group_count').to_i + next_row = result.find { _1['counts_position'].to_i == page * limit + 1 } + records = result.select do |record| + record['counts_position'] && record['counts_position'].to_i <= page * limit + end row_values = records.map { |record| row_keys.map { record[_1] } }.uniq row_depths = CountsPivotLayout.depths(row_keys, @sort) column_keys = axes.columns.each_index.map { "a#{ axes.rows.length + _1 }" } @@ -162,11 +167,11 @@ class CountsTableQuery end totals = CountsPivotTotals.new(population, axes, row_headers, column_headers).call { group_count: total, - visible_output_count: output_row_count(prefix, row_keys, row_depths, total), + visible_output_count: axes.rows.empty? ? 1 : total, column_values:, row_headers:, column_headers:, pivot_totals: totals, cells: records.map do |row| { values: axes.all.each_index.map { row["a#{ _1 }"] }, count: row['count'].to_i } - end, axis_labels: axis_labels(records, column_values) } + end, axis_labels: axis_labels(records, column_values), axis_names: @axis_names } end def details @@ -181,7 +186,7 @@ class CountsTableQuery population.timestamp(@params['at']) || population.at values = CountsVersionRepr.many(rows, resource: @definition.resource, language_code: locale.language_code, - at: point) + at: point, history_flg: population.history) values.each_with_index { |value, index| value[:event_type] = rows[index].event_type } else users = User.where(id: rows.map(&:created_by_user_id)).pluck(:id, :name).to_h @@ -198,9 +203,10 @@ class CountsTableQuery { versions: values } end - def each_page + def each_page(before_page: nil) total = nil loop do + before_page&.call result = call total ||= mode == 'summary' ? result[:group_count] : result[:count] yield result @@ -212,40 +218,50 @@ class CountsTableQuery private - def output_row_count(prefix, row_keys, depths, total) - # With no row axes the table has exactly one body row, including the zero-count case. - return 1 if row_keys.empty? + def unrestricted_population + @unrestricted_population ||= CountsPopulation.new( + @params.merge('execution_at' => population.at.iso8601(6)), authorisation_flg: false) + end - subtotals = depths.map do |depth| - keys = row_keys.first(depth).join(', ') - "(SELECT COUNT(*) FROM (SELECT #{ keys } FROM cells GROUP BY #{ keys }) subtotal_rows)" + def record_counts + return @record_counts if @record_counts + + if HistoryAccess.admin? + total = population.scope.except(:select) + .optimizer_hints('MAX_EXECUTION_TIME(10000)').count + return @record_counts = { total:, visible: total, denied: 0 } end - expressions = [total.to_s, '1'] + subtotals - @connection.select_value(<<~SQL).to_i - #{ prefix } - SELECT /*+ MAX_EXECUTION_TIME(10000) */ #{ expressions.join(' + ') } AS output_rows - SQL + total, denied = unrestricted_population.scope.except(:select) + .optimizer_hints('MAX_EXECUTION_TIME(10000)').pick(Arel.sql('COUNT(*)'), + Arel.sql("COALESCE(SUM(CASE WHEN #{ HistoryAccess.sql(@definition.model, 'v') } " + + 'THEN 0 ELSE 1 END), 0)')) + # Related/previous-version predicates may differ under public authority. + shared = QueryState.records(@params, 'conditions').empty? && + QueryState.records(@params, 'drill').none? { _1['field'] == 'changed_field' } + visible = if shared + total.to_i - denied.to_i + else + population.scope.except(:select) + .optimizer_hints('MAX_EXECUTION_TIME(10000)').count + end + @record_counts = { total: total.to_i, denied: denied.to_i, visible: } end def output_totals(visible_rows) return @output_totals if @output_totals - if HistoryAccess.admin? || @population_counts[:excluded_count].zero? + if HistoryAccess.admin? return @output_totals = { output_count: visible_rows, excluded_count: 0 } end # Only numerical aggregates leave this population. Labels, snapshots and # detail pagination continue to use the authorised population exclusively. - source = CountsPopulation.new( - @params.merge('execution_at' => population.at.iso8601(6)), authorisation_flg: false) - denied = source.scope.except(:select) - .where(Arel.sql("NOT (#{ HistoryAccess.sql(@definition.model, 'v') })")) - .optimizer_hints('MAX_EXECUTION_TIME(10000)').count + source = unrestricted_population rows = if mode == 'details' - source.scope.except(:select).optimizer_hints('MAX_EXECUTION_TIME(10000)').count + record_counts[:total] else summary_output_count(source) end - @output_totals = { output_count: rows, excluded_count: denied } + @output_totals = { output_count: rows, excluded_count: record_counts[:denied] } end def summary_output_count(source) @@ -263,13 +279,12 @@ class CountsTableQuery sql = relation.except(:select).select(Arel.sql(selected.join(', '))) .distinct.to_sql prefix = "WITH cells AS (#{ sql })" - groups = @connection.select_value(<<~SQL).to_i + @connection.select_value(<<~SQL).to_i #{ prefix } SELECT /*+ MAX_EXECUTION_TIME(10000) */ COUNT(*) FROM ( SELECT #{ keys.join(', ') } FROM cells GROUP BY #{ keys.join(', ') } ) output_groups SQL - output_row_count(prefix, keys, CountsPivotLayout.depths(keys, @sort), groups) end def locale = @locale ||= Locale.find_by!(code: @params.fetch('locale', 'ja')) @@ -345,6 +360,8 @@ class CountsTableQuery def detail_expression(key) return "(#{ population.latest_sql })" if key == 'latest' + return @definition.name_expression(key) if @definition.resource == 'tags' && + @definition.name_column(key) aliases = { 'tags' => 'tags_json', 'parents' => @definition.resource == 'posts' ? 'parent_post_ids' : 'parent_tag_ids', 'deprecated' => 'deprecated_at' } field = aliases.fetch(key, key) @@ -363,6 +380,7 @@ class CountsTableQuery def axis_labels(records, column_values) result = {} + @axis_names = {} axes.all.each_with_index do |axis, index| values = records.map { _1["a#{ index }"] } if index >= axes.rows.length @@ -375,7 +393,9 @@ class CountsTableQuery when 'language_code', 'source_language_code' Language.where(code: values).pluck(:code, :name).to_h when 'tags' - tag_labels(values) + labels = tag_labels(values) + @axis_names["a#{ index }"] = @tag_names + labels else {} end result["a#{ index }"] = labels @@ -386,20 +406,43 @@ class CountsTableQuery def tag_labels(values) internal = values.grep(/\Ainternal:/).map { _1.split(':').last.to_i } external = values.grep(/\Aexternal:/).map { _1.split(':').last.to_i } - point = population.timestamp(@params['at']) || population.at + point = population.history ? population.at : + population.timestamp(@params['at']) || population.at labels = {} + @tag_names = {} sources = [[TagVersion, internal, 'internal'], [NicoTagVersion, external, 'external']] sources.each do |model, ids, kind| next if ids.empty? table = model.table_name - model.where(tag_id: ids).where('created_at <= ?', point) - .where("NOT EXISTS (SELECT 1 FROM #{ table } newer " + - "WHERE newer.tag_id = #{ table }.tag_id AND newer.created_at <= ? " + - "AND newer.version_no > #{ table }.version_no)", point).each do |row| - names = row.respond_to?(:names) ? row.names : nil - name = names&.dig(locale.language_code, 'name') || row.name - labels["#{ kind }:#{ row.tag_id }"] = name + scope = HistoryAccess.visible(model.where(tag_id: ids)).where('created_at <= ?', point) + bound = @connection.quote(point) + if population.history && @definition.resource == 'posts' + key = kind == 'internal' ? 'tag_id' : 'external_tag_id' + pairs = population.scope.except(:select, :order).joins( + "JOIN JSON_TABLE(v.tags_json, '$[*]' COLUMNS (tag_id BIGINT PATH '$.#{ key }', " + + "version_no INT PATH '$.version_no' NULL ON EMPTY)) recorded_tag") + .where('recorded_tag.tag_id IN (?)', ids).group('recorded_tag.tag_id') + value = kind == 'internal' ? 'MAX(recorded_tag.version_no)' : 'MAX(v.created_at)' + pairs = pairs.select(Arel.sql("recorded_tag.tag_id, #{ value } AS reference_point")) + scope = scope.joins("JOIN (#{ pairs.to_sql }) recorded " + + "ON recorded.tag_id = #{ table }.tag_id") + if kind == 'internal' + scope = scope.where("#{ table }.version_no = recorded.reference_point") + else + scope = scope.where("#{ table }.created_at <= recorded.reference_point") + bound = 'recorded.reference_point' + end + end + unless population.history && kind == 'internal' && @definition.resource == 'posts' + scope = scope.where("NOT EXISTS (SELECT 1 FROM #{ table } newer " + + "WHERE newer.tag_id = #{ table }.tag_id AND newer.created_at <= #{ bound } " + + "AND newer.version_no > #{ table }.version_no)") + end + scope.each do |row| + id = "#{ kind }:#{ row.tag_id }" + labels[id] = row.name + @tag_names[id] = row.respond_to?(:names) ? row.names : {} end end labels diff --git a/backend/app/services/post_tag_search.rb b/backend/app/services/post_tag_search.rb index 7d05024..cbff09f 100644 --- a/backend/app/services/post_tag_search.rb +++ b/backend/app/services/post_tag_search.rb @@ -24,7 +24,7 @@ class PostTagSearch reference = references&.[](index) name = token.sub(/\Anot:/i, '') - name = TagName.canonicalise(locale, name).first unless reference + name = TagName.canonicalise(locale, name).first unless reference || locale.nil? negative = reference&.key?('negative') ? reference['negative'] : token.downcase.start_with?('not:') { name:, negative:, reference: } diff --git a/backend/lib/tasks/counts_exports.rake b/backend/lib/tasks/counts_exports.rake index 4c4668f..1a4e659 100644 --- a/backend/lib/tasks/counts_exports.rake +++ b/backend/lib/tasks/counts_exports.rake @@ -3,6 +3,10 @@ namespace :counts do task export_worker: :environment do stopping = false ['TERM', 'INT'].each { |signal| Signal.trap(signal) { stopping = true } } + at_exit do + path = CountsExportJobs.directory.join("#{ CountsExportJobs.worker_id }.json") + File.unlink(path) if path.exist? + end until stopping worked = Rails.application.executor.wrap { CountsExportJobs.work_one } sleep(1) unless worked diff --git a/docs/counts-exports.md b/docs/counts-exports.md index c878fb0..c3ae6ea 100644 --- a/docs/counts-exports.md +++ b/docs/counts-exports.md @@ -8,8 +8,18 @@ Active Job の既定のプロセス内 async アダプターは使用しない backend で `bundle exec rake counts:export_worker` を、systemd 等の監視対象として 別プロセスで起動する。Web プロセス内では起動しない。 -ワーカー未起動の間は待機状態となる。1 プロセスは同時に1ジョブを処理し、 -全体の実行枠は既存の2枠を共有する。ワーカーの起動・配備は実施していない。 +1監視プロセスは同時に1ジョブを専用の子プロセスで生成し、全体の実行枠は +既存の2枠を共有する。fork 前に DB 接続を閉じ、子が独立した接続を取得する。 +Web 内のスレッドや Active Job async では実行しない。Linux / POSIX の fork が必要。 +監視側は SQL 中も1秒間隔でジョブとワーカーの heartbeat を更新する。 + +`docs/systemd/btrc-counts-exports.service` を配備用 unit として追加している。 +Restart=always、KillMode=control-group、停止猶予150秒を指定している。 +User、WorkingDirectory、bundle が見つかる PATH を実際の配備先へ合わせる。 +`/etc/btrc-hub/counts-exports.env` に Web と同じ Rails secret・接続設定・ +COUNTS_EXPORT_DIRECTORY を設定し、同じ Unix ユーザーで読み書きできるようにする。 +OS 起動時の有効化・サービス起動は運用側で行う必要がある。 +今回 unit のインストール、有効化、本番での起動・再起動は行っていない。 Web とワーカーは同一リリース、Rails secret、アプリケーション環境、DB 接続先を 使用する。DB コネクションはワーカープロセスごとに必要。 @@ -30,13 +40,23 @@ Web とワーカーは同一リリース、Rails secret、アプリケーショ ジョブ・ファイルのトークンは作成元のセッションに紐付く。 既存の BAN・CSRF 制御を使用する。フロントは約800ms間隔で状態を取得する。 明細の総数は同じ読み取りトランザクション内の SQL 件数を使用する。 -集計結果の出力セル総数は生成終了まで確定させず、不確定進捗として扱う。 +queued と running に加え、running 内の phase で aggregating（集計中）と +writing（出力中）を区別する。最初の SQL より前に aggregating を記録する。 +heartbeat_at は監視側の生存時刻、updated_at は生成処理が状態を書いた時刻であり、 +処理件数と分離する。SQL 中に件数は増やさない。 +集計の総出力数は生成終了まで未確定とし、不確定進捗として扱う。 +processed は CSV では実際に生成したデータ行数、JSON のクロス集計では +実際に生成した pivot_cell 数。ヘッダーは含めず、小計・合計の物理レコードは含む。 +通常の集計行だけを数える logical_count_unit=regular_result_rows と区別する。 +出力中は約250ms間隔と最終レコードで実測件数を保存する。 完成状態はファイルの生成完了であり、ブラウザの保存完了を意味しない。 ## 取消・障害・制限 通常は約250ms間隔の生成チェックポイントで取消を検知する。 -DB 問い合わせ・ファイル I/O がブロックしている間は即時に中断できない。 +監視側も取消を確認し、子へ TERM、5秒後も終了していなければ KILL を送る。 +子の回収後に一時ファイルを削除する。DB 側の実行停止は切断検知等に依存するため、 +取消 API の受理と DB の即時停止は同義ではない。 取消済みへの遷移は生成処理の終了と一時ファイル削除後に行う。 既に始まった HTTP 転送やブラウザ側の保存は取り消せない。 @@ -45,9 +65,43 @@ DB 問い合わせ・ファイル I/O がブロックしている間は即時に 原子的に書き換える状態ファイルとプロセス間ロックにより、二重処理・取消との競合を防ぐ。 既存の同時実行2、1ファイル128MiB、保存領域512MiB、生成120秒、期限10分を維持する。 -生成時間制限もチェックポイント方式であり、実行中 SQL の強制停止ではない。 +生成120秒は監視側でも強制する。ワーカー未起動なら登録 API は503を返す。 +待機中のワーカー消失、15秒以上の heartbeat 停止、150秒以上の待機は失敗とする。 +ネットワーク切断時も UI が無期限に処理中にならないよう、状態取得は1回15秒で中断し、 +30秒以上状態を取得できなければ失敗表示へ移る。処理件数を捏造しない。 +画面からの通信断だけではサーバーの取消完了を保証しない。サーバー側の期限は独立して働く。 期限切れのジョブ・ファイル・残存状態書込ファイルは登録時またはワーカーの巡回時に削除する。 ワーカーと Web の両方を停止した間は巡回削除も停止する。 これらの処理は静的確認のみで、実環境のキュー処理・取消・転送・共有ボリュームでの 排他動作は未確認。 + +## Counts の件数と性能 + +通常の Counts 応答では、使用していない全履歴の population_count は計算しない。 +キーは null、population_count_computed は false とし、0に置き換えない。 +現在条件の output_count は権限適用前、visible_output_count は権限適用後の +通常の集計行数（明細では明細行数）。小計・合計の物理行は含めない。 +CSV の物理行と JSON の pivot_cell には集計レコードを含むため、両者を混同しない。 + +除票境界・参照禁止集合は GROUP BY で識別キー単位にまとめる。 +投稿の参照判定は、全除票履歴を JSON_CONTAINS / FIND_IN_SET で反復照合せず、 +保存済みスナップショットから参照IDを JSON_TABLE で展開して照合する。 +権限条件付きの母集団件数は同一 export インスタンス内で再利用する。 +多段小計は、認可済み母集団の CTE を MySQL の無制限 LIMIT で実体化させ、 +各枝が同じ認可判定をやり直すことを避ける。計数は COUNT(DISTINCT v.id) のまま。 + +スキーマ上、以下の追加インデックスが候補となる。migration は作成・実行していない。 + +- post_versions: (event_type, post_id, version_no) +- tag_versions / nico_tag_versions: (event_type, tag_id, version_no) +- tag_description_versions: (event_type, tag_id, language_code, version_no) +- tag_group_versions: (event_type, tag_group_id, version_no) +- material_versions: (event_type, material_id, version_no) +- 各履歴テーブル: (event_type, id)。ファイル認可の policy revision 用 MAX(id) の候補。 +- post_versions 等で未整備の (created_at)。履歴の異動期間を絞る場合の候補。 +- post_versions: (post_id, created_at, version_no)。日時指定の最新版探索の候補。 + +追加は許可後の新規 add_index migration として行い、既存 migration は変更しない。 +実DBでのインデックス容量・作成時ロック・実行計画・時間は未確認。 +索引追加なしでの最大データ量に対する応答時間は保証しない。 diff --git a/docs/systemd/btrc-counts-exports.service b/docs/systemd/btrc-counts-exports.service new file mode 100644 index 0000000..9a879b5 --- /dev/null +++ b/docs/systemd/btrc-counts-exports.service @@ -0,0 +1,21 @@ +[Unit] +Description=BTRC Hub Counts export supervisor +Wants=network-online.target +After=network-online.target +StartLimitIntervalSec=0 + +[Service] +Type=simple +User=miteruzo +WorkingDirectory=/home/miteruzo/projects/btrc-hub/backend +Environment=RAILS_ENV=production +EnvironmentFile=/etc/btrc-hub/counts-exports.env +ExecStart=/usr/bin/env bundle exec rake counts:export_worker +Restart=always +RestartSec=3 +KillMode=control-group +TimeoutStopSec=150 +UMask=0077 + +[Install] +WantedBy=multi-user.target diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 9f65d71..a966536 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -18,6 +18,13 @@ - 秘匿は React / CSS では実装しない。認可済み API のみを使う。 認証・権限変更時は取得中リクエストと履歴キャッシュを破棄し、SSR の管理者データを共有しない。 +## Mandatory Counts data contract + +- Counts のロカール非依存・内表行集合保持はルート AGENTS.md の必須契約に従う。 + 内表行の表示フィルターを作らない。言語名列はサーバーの Locale 定義を使い、 + 未登録列に別言語名を代入しない。列を隠しても取得済みの全言語データは保持する。 + ソート・コピー・JSON の投影には安定したキーと値を使い、UI 訳語を識別に使わない。 + ## Mandatory naming contract - ルート `AGENTS.md` の必須命名規約を適用する。非 callable の自前識別子に diff --git a/frontend/src/components/common/SortableColumns.tsx b/frontend/src/components/common/SortableColumns.tsx index f08837b..632f218 100644 --- a/frontend/src/components/common/SortableColumns.tsx +++ b/frontend/src/components/common/SortableColumns.tsx @@ -14,10 +14,8 @@ const SortableColumns = ({ ids, fixed = [], name, onChange, children }: { ids: string[]; fixed?: string[]; name: (id: string) => { text: string; language: string } onChange: (ids: string[]) => void; children: (ids: string[]) => ReactNode }, ) => { - const [preview, setPreview] = useState (null) const [active, setActive] = useState (null) - const latest = useRef (null) - const lastOver = useRef (null) + const drag = useRef<{ id: string; ids: string[] } | null> (null) const motion = useClientMotion () const sensors = useSensors ( useSensor (MouseSensor, { activationConstraint: { distance: 4 } }), @@ -27,12 +25,11 @@ const SortableColumns = ({ ids, fixed = [], name, onChange, children }: { if (fixed.includes (from) || fixed.includes (to)) return current const source = current.indexOf (from) const target = current.indexOf (to) - return source < 0 || target < 0 ? current : arrayMove (current, source, target) + return source < 0 || target < 0 || source === target + ? current : arrayMove (current, source, target) } const reset = () => { - latest.current = null - lastOver.current = null - setPreview (null) + drag.current = null setActive (null) } return undefined }, screenReaderInstructions: { draggable: active == null ? '' : name (active).text } }} onDragStart={event => { - latest.current = ids - setPreview (ids) - setActive (String (event.active.id)) - }} onDragOver={event => { - if (event.over == null || event.active.id === event.over.id) return - const target = String (event.over.id) - if (target === lastOver.current) return - latest.current = move (latest.current ?? ids, String (event.active.id), target) - lastOver.current = target - setPreview (latest.current) + const id = String (event.active.id) + drag.current = { id, ids } + setActive (id) }} onDragCancel={reset} onDragEnd={event => { - if (event.over != null) - { - const target = String (event.over.id) - const current = latest.current ?? ids - onChange (target === lastOver.current || target === String (event.active.id) - ? current : move (current, String (event.active.id), target)) - } + const session = drag.current reset () + if (session == null || event.over == null || session.id !== String (event.active.id) + || session.ids.length !== ids.length + || session.ids.some ((id, index) => id !== ids[index])) return + const next = move (ids, session.id, String (event.over.id)) + if (next.some ((id, index) => id !== ids[index])) onChange (next) }}> - - {children (preview ?? ids)} + + {children (ids)} {typeof document !== 'undefined' && createPortal (
{handle} diff --git a/frontend/src/components/common/SortableList.tsx b/frontend/src/components/common/SortableList.tsx index d57d19d..7564d5a 100644 --- a/frontend/src/components/common/SortableList.tsx +++ b/frontend/src/components/common/SortableList.tsx @@ -15,6 +15,12 @@ type Groups = Record type Name = { text: string; language: string } +const sameGroups = (left: Groups, right: Groups): boolean => + Object.keys (left).length === Object.keys (right).length + && Object.keys (left).every (key => left[key].length === right[key]?.length + && left[key].every ((id, index) => id === right[key][index])) + + const Group = ({ id, children }: { id: string; children: ReactNode }) => { const drop = useDroppable ({ id: `group:${id}` }) return
ReactNode onChange: (groups: Groups) => void; fixed?: string[] }, ) => { - const [preview, setPreview] = useState (null) - const previewRef = useRef (null) - const lastOver = useRef (null) + const drag = useRef<{ id: string; groups: Groups } | null> (null) const scope = useId () const [active, setActive] = useState (null) const motion = useClientMotion () - const current = preview ?? groups const sensors = useSensors ( useSensor (MouseSensor, { activationConstraint: { distance: 4 } }), useSensor (TouchSensor, { activationConstraint: { delay: 350, tolerance: 8 } }), @@ -45,7 +48,8 @@ const SortableList = ({ groups, name, renderItem, renderGroup, onChange, fixed = const relocate = (value: Groups, id: string, over: string): Groups => { const source = owner (id, value) const target = over.startsWith ('group:') ? over.slice (6) : owner (over, value) - if (source == null || target == null || fixed.includes (id) || fixed.includes (over)) + if (source == null || target == null || value[target] == null + || fixed.includes (id) || fixed.includes (over)) return value if (source === target) { @@ -62,19 +66,20 @@ const SortableList = ({ groups, name, renderItem, renderGroup, onChange, fixed = [target]: destination } } const moveGroup = (id: string, direction: number) => { + if (drag.current != null) return const keys = Object.keys (groups) const group = owner (id, groups) const target = keys[keys.indexOf (group ?? '') + direction] if (target != null) { - onChange (relocate (groups, id, `group:${target}`)) + const next = relocate (groups, id, `group:${target}`) + if (sameGroups (groups, next)) return + onChange (next) requestAnimationFrame (() => document.getElementById (`${scope}-${id}`)?.focus ()) } } const reset = () => { - previewRef.current = null - lastOver.current = null - setPreview (null) + drag.current = null setActive (null) } return undefined }, screenReaderInstructions: { draggable: active == null ? '' : name (active).text } }} onDragStart={event => { - previewRef.current = groups - lastOver.current = null - setPreview (groups) - setActive (String (event.active.id)) - }} onDragOver={event => { - if (event.over == null || event.active.id === event.over.id) return - const target = String (event.over.id) - if (lastOver.current === target) return - const next = relocate (previewRef.current ?? groups, String (event.active.id), target) - previewRef.current = next - lastOver.current = target - setPreview (next) + const id = String (event.active.id) + drag.current = { id, groups } + setActive (id) }} onDragCancel={reset} onDragEnd={event => { - if (event.over != null) - { - const id = String (event.active.id) - const target = String (event.over.id) - const latest = previewRef.current ?? groups - const next = (target === id || target === lastOver.current) - ? latest : relocate (latest, id, target) - onChange (next) - } + const session = drag.current reset () + if (session == null || event.over == null || session.id !== String (event.active.id) + || !sameGroups (session.groups, groups)) return + const next = relocate (groups, session.id, String (event.over.id)) + if (!sameGroups (groups, next)) onChange (next) }}> - {Object.entries (current).map (([group, ids]) =>
+ {/* Keep indexes and droppable geometry stable until the drop is committed. */} + {Object.entries (groups).map (([group, ids]) =>
{renderGroup (group, {ids.map (id => (fields.find (item => item.key === key)?.labels - ?? [countsFieldLabel (key)]).map (key => message (key).text).join (' › ') + ?? [countsFieldLabel (key)]).map (label => message ( + label, fields.find (item => item.key === key)?.labelValues).text).join (' › ') const [condition, setCondition] = useState (() => { const next = { ...initial, field: initial?.field ?? 'resource_id', operator: initial?.operator ?? 'eq' } @@ -39,9 +40,9 @@ const ConditionEditor = ( controls.setActions ([{ label: label (initial == null ? 'add' : 'shared:actions.edit'), onSelect: () => save (condition) }]) }, [controls, condition, initial, label, save]) - return
+ return
- }
} @@ -98,7 +100,8 @@ const CountsAdditionalConditions = ({ params, update }: { const metadata = definition.filterFields ?? definition.fields const fields = metadata.filter (field => field.operators.length > 0) const fieldName = (key: string) => (metadata.find (item => item.key === key)?.labels - ?? [countsFieldLabel (key)]).map (key => message (key).text).join (' › ') + ?? [countsFieldLabel (key)]).map (label => message ( + label, metadata.find (item => item.key === key)?.labelValues).text).join (' › ') const write = (next: CountsCondition[]) => update ( 'conditions', encodeCountState ('conditions', next)) const replace = (index: number, condition: CountsCondition) => write ( @@ -112,6 +115,7 @@ const CountsAdditionalConditions = ({ params, update }: { else replace (index, condition) } void dialogue.form ({ title: label ('additionalConditions'), footerLayout: 'inline', + size: 'large', body: controls => }) } @@ -124,7 +128,7 @@ const CountsAdditionalConditions = ({ params, update }: { const options: string[] = type === 'boolean' ? ['true', 'false'] : [...countsEventTypes] const namespace = type === 'boolean' ? 'truthValues' : 'events' return } if (type === 'datetime') - return patch (index, { [key]: value ?? '' })}/> return patch (index, { [key]: event.target.value })}/> } @@ -150,8 +154,8 @@ const CountsAdditionalConditions = ({ params, update }: { {conditions.map ((condition, index) => ( -
+