このコミットが含まれているのは:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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?
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -13,6 +13,7 @@ export type CountsExportHandle = { cancel: () => Promise<boolean> }
|
||||
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<string | null> (null)
|
||||
const creation = useRef<Promise<{ job: string }> | null> (null)
|
||||
const downloaded = useRef (false)
|
||||
const attempt = useRef<string | null> (null)
|
||||
const downloadDeadline = useRef (0)
|
||||
const releaseDownload = useRef<(() => void) | null> (null)
|
||||
const [downloadError, setDownloadError] = useState<string | null> (null)
|
||||
const pollRequest = useRef<AbortController | null> (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<ExportStatus> (
|
||||
`/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 })}</p>
|
||||
{downloadError != null && <p role="alert">
|
||||
{label (`downloadErrors.${downloadError}`)}</p>}
|
||||
{cancelError && <p role="alert">{label ('common:fetchError')}</p>}
|
||||
</div>
|
||||
}
|
||||
|
||||
@@ -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 () }
|
||||
}
|
||||
|
||||
新しいイシューから参照
ユーザーをブロックする