diff --git a/backend/app/controllers/counts_exports_controller.rb b/backend/app/controllers/counts_exports_controller.rb index 0150ac8..1a76464 100644 --- a/backend/app/controllers/counts_exports_controller.rb +++ b/backend/app/controllers/counts_exports_controller.rb @@ -24,6 +24,14 @@ class CountsExportsController < ApplicationController job_state(cancel: true) end + def start_download + result = CountsExportJobs.download_attempt(params[:token], session[:counts_export_owner]) + return head :not_found unless result + + response.headers['Cache-Control'] = 'private, no-store' + render json: result + end + def prepare locale = resolve_locale! payload = request.request_parameters.merge(request.query_parameters) @@ -45,6 +53,11 @@ class CountsExportsController < ApplicationController end def download + if params[:job].present? || params[:attempt].present? + return head :method_not_allowed unless request.get? + + return confirmed_download + end file = CountsExportDownload.find(params[:token], session[:counts_export_owner]) return head :not_found unless file @@ -55,6 +68,47 @@ class CountsExportsController < ApplicationController disposition: 'attachment' end + def confirmed_download + owner = session[:counts_export_owner] + arguments = { file_token: params[:token], attempt: params[:attempt] } + return head :not_found unless arguments[:attempt].present? && + CountsExportJobs.download_attempt(params[:job], owner, **arguments) + + file = CountsExportDownload.find(params[:token], owner) + unless file + CountsExportJobs.download_attempt(params[:job], owner, **arguments, status: 'unavailable') + return head :not_found + end + # Open before committing the response, but never buffer the entire export. + stream = File.open(file.fetch('path'), 'rb') + response.headers['Cache-Control'] = 'private, no-store' + send_file_headers! filename: "counts-#{ file.fetch('resource') }.#{ file.fetch('format') }", + type: file['format'] == 'json' ? 'application/json' : 'text/csv', + disposition: 'attachment' + job = params[:job] + authority = HistoryAccess.user + self.response_body = Enumerator.new do |output| + begin + first = true + while (chunk = stream.read(64 * 1024)) + output << chunk + if first + HistoryAccess.set(user: authority) do + CountsExportJobs.download_attempt(job, owner, **arguments, status: 'started') + end + first = false + end + end + ensure + stream.close + end + end + rescue IOError, SystemCallError + CountsExportJobs.download_attempt(params[:job], owner, **arguments, status: 'failed') + head :service_unavailable + end + private :confirmed_download + def job_state(cancel:) result = CountsExportJobs.status(params[:token], session[:counts_export_owner], cancel:) return head :not_found unless result diff --git a/backend/app/services/counts_axes.rb b/backend/app/services/counts_axes.rb index a5ed633..5ec084b 100644 --- a/backend/app/services/counts_axes.rb +++ b/backend/app/services/counts_axes.rb @@ -1,4 +1,10 @@ class CountsAxes + MAX_AXES = 8 + MASKED_VALUE = '__counts_private__'.freeze + PUBLIC_FIELDS = ['created_at', 'event_type'].freeze + class TooManyAxes < ArgumentError; end + class PrivateCondition < ArgumentError; end + GRAINS = ['year', 'quarter', 'month', 'week', 'day', 'hour'].freeze def self.time_expression(expression, grain, timezone) @@ -36,13 +42,14 @@ class CountsAxes @rows = parse(population.params.fetch('rows', '')) @columns = parse(population.params.fetch('cols', '')) @all = rows + columns - raise ArgumentError if all.length > 8 || all.map { _1['field'] }.uniq.length != all.length + raise TooManyAxes if all.length > MAX_AXES + raise ArgumentError if all.map { _1['field'] }.uniq.length != all.length end def expressions @expressions ||= all.map do |axis| field = axis.fetch('field') - case field + expression = case field when 'tags' "CASE WHEN tag_entry.tag_id IS NOT NULL " + "THEN CONCAT('internal:', tag_entry.tag_id) " + @@ -53,6 +60,11 @@ class CountsAxes @definition.fields[field] == :datetime ? self.class.time_expression( expression, axis.fetch('grain', 'month'), @population.timezone) : expression end + if !HistoryAccess.admin? && !PUBLIC_FIELDS.include?(field) + expression = "CASE WHEN #{ HistoryAccess.sql(@definition.model, 'v') } " + + "THEN CAST((#{ expression }) AS CHAR) ELSE '#{ MASKED_VALUE }' END" + end + expression end end @@ -73,6 +85,9 @@ class CountsAxes condition = @definition.changes.map do |field| "(changed.field_key = '#{ field }' AND #{ @population.changed_sql(field) })" end.join(' OR ') + unless HistoryAccess.admin? + condition = "(#{ condition }) OR NOT (#{ HistoryAccess.sql(@definition.model, 'v') })" + end result += " JOIN (#{ fields }) changed ON (#{ condition })" end result diff --git a/backend/app/services/counts_export_jobs.rb b/backend/app/services/counts_export_jobs.rb index b3536f0..34b1ff8 100644 --- a/backend/app/services/counts_export_jobs.rb +++ b/backend/app/services/counts_export_jobs.rb @@ -184,7 +184,29 @@ class CountsExportJobs write(state) end state.slice('status', 'phase', 'processed', 'total', 'format', 'token', 'error', - 'updated_at', 'heartbeat_at') + 'updated_at', 'heartbeat_at', 'download_attempt', 'download_status') + end + end + + def self.download_attempt(token, owner, file_token: nil, attempt: nil, status: nil) + id = authorised_id(token, owner) + return unless id + + synchronise do + state = read(id) + return unless (state && state['owner'] == owner && state['status'] == 'ready' && + state['expires_at'] > Time.current.to_f && HistoryAccess.authorised_artifact?(state)) + if attempt + return unless state['download_attempt'] == attempt && state['token'] == file_token + return unless state['download_status'] == 'pending' + + state['download_status'] = status if status + else + state['download_attempt'] = SecureRandom.hex(24) + state['download_status'] = 'pending' + end + write(state) + state.slice('download_attempt', 'download_status', 'token') end end diff --git a/backend/app/services/counts_pivot_totals.rb b/backend/app/services/counts_pivot_totals.rb index 44c5978..f25a0ab 100644 --- a/backend/app/services/counts_pivot_totals.rb +++ b/backend/app/services/counts_pivot_totals.rb @@ -19,7 +19,9 @@ class CountsPivotTotals (0...column_depth).map { @axes.rows.length + _1 } expressions = indexes.map { @axes.expressions[_1] } scope = @population.scope.except(:select, :order, :where, :from).from('authorised v') - joins = @axes.joins(indexes.map { @axes.all[_1].fetch('field') }) + additive = @axes.rows.any? { _1['field'] == 'tags' } + retained = additive ? (0...@axes.rows.length).to_a | indexes : indexes + joins = @axes.joins(retained.map { @axes.all[_1].fetch('field') }) scope = scope.joins(joins) unless joins.empty? if row_depth.positive? tuples = headers.map { _1[:values] }.uniq @@ -42,9 +44,23 @@ class CountsPivotTotals padding = (expressions.length...@axes.all.length).map { "NULL AS v#{ _1 }" } fields = ["#{ row_depth } AS row_depth", "#{ column_depth } AS column_depth"] + grouped + padding + ['COUNT(DISTINCT v.id) AS count'] - scope = scope.group(*expressions.map { Arel.sql(_1) }) if expressions.any? - scope.optimizer_hints('MAX_EXECUTION_TIME(10000)') - .select(Arel.sql(fields.join(', '))).to_sql + if additive + # Sum full row tuples, not page cells; preserve DISTINCT inside each tuple. + leaf = retained.map { @axes.expressions[_1] } + selected = leaf.each_with_index.map { |expression, index| "#{ expression } AS k#{ index }" } + inner = scope.group(*leaf.map { Arel.sql(_1) }) + .select(Arel.sql((selected + ['COUNT(DISTINCT v.id) AS count']).join(', '))).to_sql + coordinates = indexes.map { "k#{ retained.index(_1) }" } + projected = coordinates.each_with_index.map { |key, index| "#{ key } AS v#{ index }" } + output = ["#{ row_depth } AS row_depth", "#{ column_depth } AS column_depth"] + + projected + padding + ['SUM(count) AS count'] + grouping = coordinates.empty? ? '' : " GROUP BY #{ coordinates.join(', ') }" + "SELECT #{ output.join(', ') } FROM (#{ inner }) leaf#{ grouping }" + else + scope = scope.group(*expressions.map { Arel.sql(_1) }) if expressions.any? + scope.optimizer_hints('MAX_EXECUTION_TIME(10000)') + .select(Arel.sql(fields.join(', '))).to_sql + end end end return [] if queries.empty? diff --git a/backend/app/services/counts_table_query.rb b/backend/app/services/counts_table_query.rb index 49bb3fe..bf4ee40 100644 --- a/backend/app/services/counts_table_query.rb +++ b/backend/app/services/counts_table_query.rb @@ -16,7 +16,9 @@ class CountsTableQuery params = CountsDefaults.for(params).slice(*KEYS).merge(params) @params = params - @population = CountsPopulation.new(params) + @mode = params.fetch('mode', 'summary') + validate_summary_conditions! + @population = CountsPopulation.new(params, authorisation_flg: mode != 'summary') @definition = population.definition @connection = @definition.model.connection @axes = CountsAxes.new(population) @@ -64,6 +66,7 @@ class CountsTableQuery definition: population.history ? 'history' : 'latest', timezone: population.timezone.identifier, rows: axes.rows, cols: axes.columns, fields: @definition.metadata, + masked_value: HistoryAccess.admin? ? nil : CountsAxes::MASKED_VALUE, count: record_counts[:visible] } end @@ -218,19 +221,41 @@ class CountsTableQuery private + # Public timeline predicates may include hidden records. Private predicates must + # never become a count oracle, even when the selected axes themselves are public. + def validate_summary_conditions! + return if mode != 'summary' || HistoryAccess.admin? + + fields = QueryState.records(@params, 'conditions') + QueryState.records(@params, 'drill') + private_fields = fields.any? { !CountsAxes::PUBLIC_FIELDS.include?(_1['field']) } + private_parameters = ['tags', 'tag_refs', 'tag', 'kind', 'resource_id', 'user_id', + 'name_scope'].any? do |key| + @params.any? { |name, value| (name == key || name.start_with?("#{ key }.")) && value != '' } + end + raise CountsAxes::PrivateCondition if private_fields || private_parameters + + ['column_order', 'measures'].each do |key| + next if QueryState.records(@params, key).empty? + columns = QueryState.axes(@params.fetch('cols', '')) + raise CountsAxes::PrivateCondition if columns.any? do |axis| + !CountsAxes::PUBLIC_FIELDS.include?(axis['field']) + end + end + end + def unrestricted_population @unrestricted_population ||= CountsPopulation.new( @params.merge('execution_at' => population.at.iso8601(6)), authorisation_flg: false) end def summary_population - unrestricted_population + population end def record_counts return @record_counts if @record_counts - if HistoryAccess.admin? + if HistoryAccess.admin? || mode == 'summary' total = population.scope.except(:select) .optimizer_hints('MAX_EXECUTION_TIME(10000)').count return @record_counts = { total:, visible: total, denied: 0 } @@ -254,6 +279,9 @@ class CountsTableQuery def output_totals visible_rows return @output_totals if @output_totals + if mode == 'summary' + return @output_totals = { output_count: visible_rows, excluded_count: nil } + end if HistoryAccess.admin? return @output_totals = { output_count: visible_rows, excluded_count: 0 } end @@ -405,7 +433,7 @@ class CountsTableQuery if index >= axes.rows.length values += column_values.map { _1[index - axes.rows.length] } end - values = values.compact.uniq + values = values.compact.uniq - [CountsAxes::MASKED_VALUE] labels = case axis['field'] when 'user' User.where(id: values).pluck(:id, :name).to_h.transform_keys(&:to_s) diff --git a/backend/config/routes.rb b/backend/config/routes.rb index c99dc7a..62f7f3f 100644 --- a/backend/config/routes.rb +++ b/backend/config/routes.rb @@ -3,6 +3,7 @@ Rails.application.routes.draw do get 'counts/definitions/:resource/candidates', to: 'counts_tables#candidates' get 'counts/query', to: 'counts_tables#index' post 'counts/export/jobs', to: 'counts_exports#create_job' + post 'counts/export/jobs/:token/download', to: 'counts_exports#start_download' get 'counts/export/jobs/:token', to: 'counts_exports#job_status' delete 'counts/export/jobs/:token', to: 'counts_exports#cancel_job' post 'counts/export/prepare', to: 'counts_exports#prepare' diff --git a/frontend/src/components/counts/CountsExportProgress.tsx b/frontend/src/components/counts/CountsExportProgress.tsx index b9e6647..4eb143f 100644 --- a/frontend/src/components/counts/CountsExportProgress.tsx +++ b/frontend/src/components/counts/CountsExportProgress.tsx @@ -13,6 +13,7 @@ export type CountsExportHandle = { cancel: () => Promise } type ExportStatus = { status: 'starting' | 'queued' | 'running' | 'cancelling' | 'cancelled' | 'ready' | 'failed' processed: number; total: number | null; token?: string + downloadAttempt?: string; downloadStatus?: string; error?: string phase?: 'queued' | 'aggregating' | 'writing'; heartbeatAt?: number; updatedAt?: number } @@ -31,10 +32,15 @@ const CountsExportProgress = ( const job = useRef (null) const creation = useRef | null> (null) const downloaded = useRef (false) + const attempt = useRef (null) + const downloadDeadline = useRef (0) + const releaseDownload = useRef<(() => void) | null> (null) + const [downloadError, setDownloadError] = useState (null) const pollRequest = useRef (null) useEffect (() => { handle.cancel = async () => { if (terminal.current) return true + releaseDownload.current?.() cancelled.current = true if (job.current != null) { @@ -81,13 +87,46 @@ const CountsExportProgress = ( if (!active) return lastResponse = Date.now () setState (result) - terminal.current = ['ready', 'failed', 'cancelled'].includes (result.status) + terminal.current = ['failed', 'cancelled'].includes (result.status) if (result.status === 'ready' && result.token != null && !downloaded.current) { downloaded.current = true - startAPIDownload ('/counts/export', result.token, () => { - if (active) setState (previous => ({ ...previous, status: 'failed' })) - }) + const receipt = await apiPost ( + `/counts/export/jobs/${encodeURIComponent (token)}/download`, {}, { + locale, signal: controller.signal }) + if (!active || cancelled.current) return + if (receipt.downloadAttempt == null) throw new Error ('Missing download attempt') + attempt.current = receipt.downloadAttempt + downloadDeadline.current = Date.now () + 30_000 + releaseDownload.current = startAPIDownload ('/counts/export', result.token, () => { + if (active) + { + terminal.current = true + setDownloadError ('failed') + } + }, new URLSearchParams ({ job: token, attempt: receipt.downloadAttempt })) + } + if (attempt.current != null && result.downloadAttempt === attempt.current) + { + if (result.downloadStatus === 'started' && !cancelled.current) + { + terminal.current = true + // Native navigation continues after the dialogue unmounts. + releaseDownload.current = null + controls.close () + return + } + if (['failed', 'unavailable'].includes (result.downloadStatus ?? '')) + { + terminal.current = true + setDownloadError (result.downloadStatus!) + } + } + if (attempt.current != null && Date.now () > downloadDeadline.current) + { + terminal.current = true + setDownloadError ('timeout') + releaseDownload.current?.() } if (!terminal.current) timer = setTimeout (() => poll (token), 800) } @@ -99,6 +138,7 @@ const CountsExportProgress = ( || Date.now () - lastResponse > 30_000) { terminal.current = true + if (downloaded.current) setDownloadError ('unavailable') setState (previous => ({ ...previous, status: 'failed' })) return } @@ -135,8 +175,9 @@ const CountsExportProgress = ( active = false if (timer != null) clearTimeout (timer) pollRequest.current?.abort () + releaseDownload.current?.() } - }, [handle, locale, payload]) + }, [handle, locale, payload, controls]) useEffect (() => { controls.setActions ([]) }, [controls]) @@ -161,6 +202,8 @@ const CountsExportProgress = ( ? label ('exportProgress.processed', { count: state.processed }) : label ('exportProgress.measured', { processed: state.processed, total: state.total ?? 0, percent })}

+ {downloadError != null &&

+ {label (`downloadErrors.${downloadError}`)}

} {cancelError &&

{label ('common:fetchError')}

} } diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 6d5861c..210d47b 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -219,7 +219,8 @@ export const startAPIDownload = ( path: string, token: string, onError?: () => void, -): void => { + query?: URLSearchParams, +): (() => void) => { const frame = document.createElement ('iframe') frame.hidden = true frame.addEventListener ('load', () => { @@ -231,8 +232,10 @@ export const startAPIDownload = ( } }) frame.addEventListener ('error', () => { onError?.(); frame.remove () }) - frame.src = `${API_BASE_URL}${path}/files/${encodeURIComponent (token)}` + const suffix = query == null ? '' : `?${query.toString ()}` + frame.src = `${API_BASE_URL}${path}/files/${encodeURIComponent (token)}${suffix}` document.body.append (frame) // Attachments do not reliably fire load; keep this isolated from page navigation. - window.setTimeout (() => frame.remove (), 10 * 60_000) + const timer = window.setTimeout (() => frame.remove (), 10 * 60_000) + return () => { window.clearTimeout (timer); frame.remove () } }