scripts/backfill_legacy_download_fail_daily_version.ts
Rama main · réplica de lectura del repositorio original.
/*
* Remove legacy download_fail from daily_version.fail.
*
* Updater plugins before 6.53.0 / 7.53.0 / 8.53.0 report a failed update check
* as download_fail against the bundle the device already runs, and the stats
* endpoint used to count it as a fail of that bundle. cron_stat_app now
* subtracts them on read, but only refreshes the current billing cycle. This script recomputes every whole day still held by Analytics
* Engine (89 days) from version_usage minus the legacy download_fail logs, the
* same correction readStatsVersionCF applies. It only lowers counts and is
* idempotent.
*
* Dry run:
* bun run admin:backfill-legacy-download-fail
*
* Apply:
* bun run admin:backfill-legacy-download-fail --apply
*
* Optional:
* --app-id=com.example.app
* --days=30
* --concurrency=4
* --env-file=./internal/cloudflare/.env.preprod
*/
import { mkdir, writeFile } from 'node:fs/promises'
import process from 'node:process'
import { escapeSqlString } from '../supabase/functions/_backend/utils/cloudflare.ts'
import { legacyDownloadFailKey, legacyDownloadFailPluginSqlCF, sumLegacyDownloadFails } from '../supabase/functions/_backend/utils/legacyDownloadFail.ts'
import { asyncPool, createSupabaseServiceClient, DEFAULT_ENV_FILE, getArgValue, getRequiredEnv, loadEnv, parsePositiveInteger } from './admin_stripe_backfill_utils.ts'
const DAY_MS = 86_400_000
const RETENTION_DAYS = 89
const DEFAULT_CONCURRENCY = 2
const MAX_RATE_LIMIT_RETRIES = 6
const DAILY_VERSION_PAGE_SIZE = 1000
const UPDATE_BATCH_SIZE = 500
const RESULT_OUTPUT = './tmp/legacy_download_fail_daily_version_backfill.json'
const DAY_BUCKET = `formatDateTime(toStartOfInterval(timestamp, INTERVAL '1' DAY), '%Y-%m-%d')`
interface AnalyticsEngineConfig {
accountId: string
token: string
}
interface Change {
app_id: string
date: string
version_name: string
before: number
after: number
}
function startOfUtcDay(date: Date) {
return new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate()))
}
function formatDate(date: Date) {
return date.toISOString().replace('T', ' ').slice(0, 19)
}
async function queryAnalyticsEngine<T>(config: AnalyticsEngineConfig, query: string): Promise<T[]> {
for (let attempt = 0; ; attempt++) {
const response = await fetch(`https://api.cloudflare.com/client/v4/accounts/${config.accountId}/analytics_engine/sql`, {
method: 'POST',
headers: { 'Authorization': `Bearer ${config.token}`, 'Content-Type': 'text/plain; charset=utf-8' },
body: query,
})
// Analytics Engine rate limits bursts of queries: back off and retry.
if (response.status === 429 && attempt < MAX_RATE_LIMIT_RETRIES) {
await response.body?.cancel()
await new Promise(resolve => setTimeout(resolve, 2 ** attempt * 1000))
continue
}
if (!response.ok)
throw new Error(`Analytics Engine query failed (${response.status}): ${await response.text()}`)
const json = await response.json() as { data?: T[] }
return json.data ?? []
}
}
async function readLegacyDownloadFails(config: AnalyticsEngineConfig, start: Date, end: Date, appId: string | null) {
const appFilter = appId ? `AND index1 = '${escapeSqlString(appId)}'` : ''
const byApp = new Map<string, Map<string, number>>()
// One week per query keeps each response small.
for (let from = start.getTime(); from < end.getTime(); from += 7 * DAY_MS) {
const to = new Date(Math.min(from + 7 * DAY_MS, end.getTime()))
const rows = await queryAnalyticsEngine<{ app_id: string, bucket: string, version_name: string, plugin_version: string, count: number | string }>(config, `SELECT
index1 AS app_id,
${DAY_BUCKET} AS bucket,
blob3 AS version_name,
blob7 AS plugin_version,
sum(_sample_interval) AS count
FROM app_log
WHERE
blob2 = 'download_fail'
AND ${legacyDownloadFailPluginSqlCF('blob7')}
AND blob10 = ''
AND timestamp >= toDateTime('${formatDate(new Date(from))}')
AND timestamp < toDateTime('${formatDate(to)}')
${appFilter}
GROUP BY app_id, bucket, version_name, plugin_version`)
const rowsByApp = Map.groupBy(rows, row => row.app_id)
for (const [app, appRows] of rowsByApp) {
const counts = byApp.get(app) ?? new Map<string, number>()
for (const [key, count] of sumLegacyDownloadFails(appRows))
counts.set(key, (counts.get(key) ?? 0) + count)
byApp.set(app, counts)
}
}
return byApp
}
async function readVersionFails(config: AnalyticsEngineConfig, appId: string, start: Date, end: Date) {
const rows = await queryAnalyticsEngine<{ date: string, version_name: string, fail: number | string }>(config, `SELECT
${DAY_BUCKET} AS date,
blob2 AS version_name,
sum(if(blob3 = 'fail', _sample_interval, 0)) AS fail
FROM version_usage
WHERE
index1 = '${escapeSqlString(appId)}'
AND timestamp >= toDateTime('${formatDate(start)}')
AND timestamp < toDateTime('${formatDate(end)}')
GROUP BY date, version_name`)
return new Map(rows.map(row => [legacyDownloadFailKey(row.date, row.version_name), Number(row.fail)]))
}
export async function main() {
const apply = process.argv.includes('--apply')
const envFile = getArgValue(process.argv, '--env-file') ?? DEFAULT_ENV_FILE
const appId = getArgValue(process.argv, '--app-id')
const days = Math.min(parsePositiveInteger(getArgValue(process.argv, '--days'), '--days', RETENTION_DAYS), RETENTION_DAYS)
const concurrency = parsePositiveInteger(getArgValue(process.argv, '--concurrency'), '--concurrency', DEFAULT_CONCURRENCY)
const env = { ...process.env, ...(await loadEnv(envFile)) }
const config = { accountId: getRequiredEnv(env, 'CF_ACCOUNT_ANALYTICS_ID'), token: getRequiredEnv(env, 'CF_ANALYTICS_TOKEN') }
const supabase = createSupabaseServiceClient(env)
// Whole UTC days only: a day cut by retention or still running would be partial.
const end = startOfUtcDay(new Date())
const start = new Date(end.getTime() - days * DAY_MS)
console.log(`${apply ? 'Applying' : 'Dry run'}: ${start.toISOString().slice(0, 10)} to ${end.toISOString().slice(0, 10)} (exclusive)${appId ? ` for ${appId}` : ''}`)
const legacyByApp = await readLegacyDownloadFails(config, start, end, appId)
console.log(`${legacyByApp.size} apps have legacy download_fail logs`)
const changes: Change[] = []
const failures: Array<{ app_id: string, error: string }> = []
await asyncPool(concurrency, [...legacyByApp.keys()], async (app) => {
try {
const legacy = legacyByApp.get(app)!
const versionFails = await readVersionFails(config, app, start, end)
for (let page = 0; ; page++) {
const { data, error } = await supabase
.from('daily_version')
.select('date, version_name, fail')
.eq('app_id', app)
.gte('date', start.toISOString().slice(0, 10))
.lt('date', end.toISOString().slice(0, 10))
.order('date')
.order('version_name')
.range(page * DAILY_VERSION_PAGE_SIZE, (page + 1) * DAILY_VERSION_PAGE_SIZE - 1)
if (error)
throw error
for (const row of data ?? []) {
const key = legacyDownloadFailKey(row.date, row.version_name)
const legacyFails = legacy.get(key)
const recordedFails = versionFails.get(key)
if (!legacyFails || recordedFails === undefined)
continue
// A stored day can hold fewer fails than Analytics Engine (last refreshed
// before the day ended): only ever lower it, so a rerun changes nothing.
const before = Number(row.fail ?? 0)
const after = Math.min(before, Math.max(0, recordedFails - legacyFails))
if (after !== before)
changes.push({ app_id: app, date: row.date, version_name: row.version_name, before, after })
}
if ((data?.length ?? 0) < DAILY_VERSION_PAGE_SIZE)
break
}
}
catch (error) {
failures.push({ app_id: app, error: error instanceof Error ? error.message : String(error) })
}
})
if (apply) {
// Every changed row exists: an upsert on the primary key only rewrites fail,
// one request per batch instead of one per row.
const batches = Array.from({ length: Math.ceil(changes.length / UPDATE_BATCH_SIZE) }, (_, index) => changes.slice(index * UPDATE_BATCH_SIZE, (index + 1) * UPDATE_BATCH_SIZE))
await asyncPool(concurrency, batches, async (batch) => {
const { error } = await supabase
.from('daily_version')
.upsert(batch.map(change => ({ app_id: change.app_id, date: change.date, version_name: change.version_name, fail: change.after })), { onConflict: 'app_id,date,version_name' })
if (error)
failures.push({ app_id: batch[0].app_id, error: `batch of ${batch.length} from ${batch[0].date}: ${error.message}` })
})
}
const removed = changes.reduce((total, change) => total + change.before - change.after, 0)
console.log(`${changes.length} daily_version rows ${apply ? 'updated' : 'to update'} across ${new Set(changes.map(change => change.app_id)).size} apps, ${removed} legacy fails removed`)
await mkdir('./tmp', { recursive: true })
await writeFile(RESULT_OUTPUT, `${JSON.stringify({ apply, start, end, changes, failures }, null, 2)}\n`)
console.log(`Details written to ${RESULT_OUTPUT}`)
if (failures.length > 0)
throw new Error(`Legacy download_fail backfill finished with ${failures.length} failures`)
}
if (import.meta.main)
await main()