From 43e5b4b3e57de51af05de84ad55638b69bb848c6 Mon Sep 17 00:00:00 2001 From: Martin Donadieu Date: Tue, 28 Jul 2026 11:06:41 +0300 Subject: [PATCH 1/2] fix(dashboard): read update delivery latency from Analytics Engine Prod plugin stats live in APP_LOG, not Postgres. Use the same CF/SB dual path as other private stats so Time to deliver an update can show data. Co-authored-by: Cursor --- .../_backend/private/update_delivery_stats.ts | 286 +++++++++++++++++- .../functions/_backend/utils/cloudflare.ts | 89 +++++- .../collectAnalyticsEngineSqlFixtures.ts | 25 +- tests/update-delivery-stats.unit.test.ts | 106 +++++++ 4 files changed, 494 insertions(+), 12 deletions(-) diff --git a/supabase/functions/_backend/private/update_delivery_stats.ts b/supabase/functions/_backend/private/update_delivery_stats.ts index ca8610179b..013bba1914 100644 --- a/supabase/functions/_backend/private/update_delivery_stats.ts +++ b/supabase/functions/_backend/private/update_delivery_stats.ts @@ -1,18 +1,23 @@ +import type { Dayjs } from 'dayjs' import type { Context } from 'hono' +import type { UpdateDeliveryTimingEventCF } from '../utils/cloudflare.ts' import type { MiddlewareKeyVariables } from '../utils/hono.ts' import dayjs from 'dayjs' import utc from 'dayjs/plugin/utc.js' import { Hono } from 'hono/tiny' +import { readUpdateDeliveryTimingEventsCF } from '../utils/cloudflare.ts' import { parseBody, simpleError, useCors } from '../utils/hono.ts' import { middlewareAuth } from '../utils/hono_jwt.ts' import { cloudlog } from '../utils/logging.ts' import { closeClient, getPgClient, logPgError } from '../utils/pg.ts' import { checkPermission } from '../utils/rbac.ts' -import { supabaseClient as useSupabaseClient } from '../utils/supabase.ts' +import { supabaseAdmin, supabaseClient as useSupabaseClient } from '../utils/supabase.ts' dayjs.extend(utc) const maxPeriodDays = 365 +const maxDeliveryMs = 7_200_000 +const pairingLookbackMs = 2 * 60 * 60 * 1000 type UpdateDeliveryPeriodDays = number type UpdateDeliveryScope = 'app' | 'org' | 'platform' @@ -41,11 +46,20 @@ interface UpdateDeliveryOverviewRow { p99_ms: number | string | null } +interface DeliverySample { + day: string + app_id: string + device_id: string + duration_ms: number +} + type NumericValue = number | string | null | undefined const endActions = ['download_complete', 'download_zip_complete'] as const const startActions = ['download_0', 'download_zip_start', 'download_manifest_start'] as const const timingActions = [...endActions, ...startActions] as const +const endActionSet = new Set(endActions) +const startActionSet = new Set(startActions) const durationExpression = String.raw`CASE WHEN s.metadata ? 'duration_ms' @@ -238,6 +252,158 @@ function toMetric(value: NumericValue, decimals = 0) { return Math.round(numeric * factor) / factor } +function parseMetaDurationMs(metadata: Record | null | undefined): number | null { + if (!metadata) + return null + for (const key of ['duration_ms', 'duration'] as const) { + const raw = metadata[key] + if (typeof raw !== 'string' || raw.length === 0 || raw.length > 15) + continue + if (!/^\d+(?:\.\d+)?$/.test(raw)) + continue + const value = Number(raw) + if (!Number.isFinite(value)) + continue + return value + } + return null +} + +function percentileCont(sorted: number[], q: number): number | null { + if (!sorted.length) + return null + if (sorted.length === 1) + return sorted[0]! + const index = (sorted.length - 1) * q + const lower = Math.floor(index) + const upper = Math.ceil(index) + if (lower === upper) + return sorted[lower]! + const weight = index - lower + return sorted[lower]! * (1 - weight) + sorted[upper]! * weight +} + +function isValidDuration(durationMs: number | null | undefined): durationMs is number { + return typeof durationMs === 'number' + && Number.isFinite(durationMs) + && durationMs >= 0 + && durationMs <= maxDeliveryMs +} + +function buildDeliveriesFromEvents( + events: UpdateDeliveryTimingEventCF[], + options: { + periodStartMs: number + allowPairing: boolean + }, +): DeliverySample[] { + const startsByKey = new Map() + if (options.allowPairing) { + for (const event of events) { + if (!startActionSet.has(event.action)) + continue + const createdAtMs = Date.parse(event.created_at) + if (!Number.isFinite(createdAtMs)) + continue + const key = `${event.app_id}\0${event.device_id}\0${event.version_name || 'unknown'}` + const list = startsByKey.get(key) + if (list) + list.push(createdAtMs) + else + startsByKey.set(key, [createdAtMs]) + } + for (const list of startsByKey.values()) + list.sort((a, b) => a - b) + } + + const deliveries: DeliverySample[] = [] + for (const event of events) { + if (!endActionSet.has(event.action)) + continue + const endMs = Date.parse(event.created_at) + if (!Number.isFinite(endMs) || endMs < options.periodStartMs) + continue + + let durationMs = parseMetaDurationMs(event.metadata) + if (!isValidDuration(durationMs) && options.allowPairing) { + const key = `${event.app_id}\0${event.device_id}\0${event.version_name || 'unknown'}` + const starts = startsByKey.get(key) + if (starts?.length) { + let matchedStart: number | null = null + for (let i = starts.length - 1; i >= 0; i -= 1) { + const startMs = starts[i]! + if (startMs > endMs) + continue + if (endMs - startMs > pairingLookbackMs) + break + matchedStart = startMs + break + } + if (matchedStart !== null) + durationMs = endMs - matchedStart + } + } + + if (!isValidDuration(durationMs)) + continue + + deliveries.push({ + day: new Date(endMs).toISOString().slice(0, 10), + app_id: event.app_id, + device_id: event.device_id, + duration_ms: durationMs, + }) + } + + return deliveries +} + +function aggregateDeliverySamples(samples: DeliverySample[]): { + dailyRows: UpdateDeliveryDailyRow[] + overviewRow: UpdateDeliveryOverviewRow +} { + const byDay = new Map() + const allDurations: number[] = [] + const devices = new Set() + + for (const sample of samples) { + allDurations.push(sample.duration_ms) + devices.add(`${sample.app_id}\0${sample.device_id}`) + const dayList = byDay.get(sample.day) + if (dayList) + dayList.push(sample.duration_ms) + else + byDay.set(sample.day, [sample.duration_ms]) + } + + allDurations.sort((a, b) => a - b) + const dailyRows: UpdateDeliveryDailyRow[] = [...byDay.entries()] + .sort(([a], [b]) => a.localeCompare(b)) + .map(([day, durations]) => { + const sorted = [...durations].sort((a, b) => a - b) + return { + day, + samples: sorted.length, + p50_ms: percentileCont(sorted, 0.5), + p75_ms: percentileCont(sorted, 0.75), + p95_ms: percentileCont(sorted, 0.95), + p99_ms: percentileCont(sorted, 0.99), + } + }) + + return { + dailyRows, + overviewRow: { + samples: allDurations.length, + devices: devices.size, + p50_ms: percentileCont(allDurations, 0.5), + p75_ms: percentileCont(allDurations, 0.75), + p95_ms: percentileCont(allDurations, 0.95), + p99_ms: percentileCont(allDurations, 0.99), + }, + } +} + function buildUpdateDeliveryResponse(input: { labels: string[] days: UpdateDeliveryPeriodDays @@ -316,16 +482,28 @@ async function assertPlatformAdmin(c: Context) { throw simpleError('not_admin', 'Not admin - only admin users can access platform delivery latency') } -async function readUpdateDeliveryStats( +async function listOrgAppIds(c: Context, orgId: string): Promise { + const { data, error } = await supabaseAdmin(c) + .from('apps') + .select('app_id') + .eq('owner_org', orgId) + if (error) { + cloudlog({ requestId: c.get('requestId'), message: 'list_org_app_ids_error', error }) + throw simpleError('fetch_error', 'Failed to fetch organization apps', { error: String(error) }) + } + return (data ?? []).map(row => row.app_id).filter((appId): appId is string => typeof appId === 'string' && appId.length > 0) +} + +async function readUpdateDeliveryStatsSB( c: Context, scope: UpdateDeliveryScope, days: UpdateDeliveryPeriodDays, - scopeId?: string, + scopeId: string | undefined, + labels: string[], + start: Dayjs, + endExclusive: Dayjs, + endInclusive: Dayjs, ) { - const endExclusive = dayjs().utc().add(1, 'day').startOf('day') - const start = endExclusive.subtract(days, 'day') - const endInclusive = endExclusive.subtract(1, 'millisecond') - const labels = generateDateLabels(start.toDate(), endExclusive.subtract(1, 'day').toDate()) const db = getPgClient(c, true) try { @@ -358,7 +536,7 @@ async function readUpdateDeliveryStats( }) } catch (error) { - logPgError(c, 'readUpdateDeliveryStats', error) + logPgError(c, 'readUpdateDeliveryStatsSB', error) throw error } finally { @@ -366,6 +544,94 @@ async function readUpdateDeliveryStats( } } +async function readUpdateDeliveryStatsCF( + c: Context, + scope: UpdateDeliveryScope, + days: UpdateDeliveryPeriodDays, + scopeId: string | undefined, + labels: string[], + start: Dayjs, + endExclusive: Dayjs, + endInclusive: Dayjs, +) { + const allowPairing = scope !== 'platform' + const queryStart = allowPairing + ? start.subtract(2, 'hour') + : start + + let appIds: string[] | undefined + if (scope === 'app') { + if (!scopeId) + throw simpleError('missing_params', 'app_id is required for app scope') + appIds = [scopeId] + } + else if (scope === 'org') { + if (!scopeId) + throw simpleError('missing_params', 'org_id is required for org scope') + appIds = await listOrgAppIds(c, scopeId) + } + + const events = await readUpdateDeliveryTimingEventsCF(c, { + start_date: queryStart.toISOString(), + end_date: endExclusive.toISOString(), + actions: allowPairing ? [...timingActions] : [...endActions], + app_ids: appIds, + }) + + const samples = buildDeliveriesFromEvents(events, { + periodStartMs: start.valueOf(), + allowPairing, + }) + const { dailyRows, overviewRow } = aggregateDeliverySamples(samples) + + return buildUpdateDeliveryResponse({ + labels, + days, + start: start.toISOString(), + end: endInclusive.toISOString(), + scope, + dailyRows, + overviewRow, + }) +} + +async function readUpdateDeliveryStats( + c: Context, + scope: UpdateDeliveryScope, + days: UpdateDeliveryPeriodDays, + scopeId?: string, +) { + const endExclusive = dayjs().utc().add(1, 'day').startOf('day') + const start = endExclusive.subtract(days, 'day') + const endInclusive = endExclusive.subtract(1, 'millisecond') + const labels = generateDateLabels(start.toDate(), endExclusive.subtract(1, 'day').toDate()) + + // Same dual-path pattern as private/stats: Analytics Engine in prod, Postgres locally. + if (c.env.APP_LOG) { + return readUpdateDeliveryStatsCF( + c, + scope, + days, + scopeId, + labels, + start, + endExclusive, + endInclusive, + ) + } + + return readUpdateDeliveryStatsSB( + c, + scope, + days, + scopeId, + labels, + start, + endExclusive, + endInclusive, + ) +} + export const app = new Hono() app.use('/', useCors) @@ -423,8 +689,12 @@ app.post('/', middlewareAuth, async (c) => { export const updateDeliveryStatsTestUtils = { buildUpdateDeliveryResponse, + buildDeliveriesFromEvents, + aggregateDeliverySamples, generateDateLabels, normalizePeriodDays, normalizeScope, + parseMetaDurationMs, + percentileCont, toMetric, } diff --git a/supabase/functions/_backend/utils/cloudflare.ts b/supabase/functions/_backend/utils/cloudflare.ts index 996f644346..89bcca8364 100644 --- a/supabase/functions/_backend/utils/cloudflare.ts +++ b/supabase/functions/_backend/utils/cloudflare.ts @@ -201,7 +201,7 @@ function serializeStatsMetadata(metadata?: StatsMetadata): string { return metadata && Object.keys(metadata).length > 0 ? JSON.stringify(metadata) : '' } -function parseStatsMetadata(metadata: unknown): StatsMetadata | null { +export function parseStatsMetadata(metadata: unknown): StatsMetadata | null { if (typeof metadata !== 'string' || metadata === '') return null @@ -1240,6 +1240,93 @@ LIMIT ${limit}` return [] as StatRowCF[] } +export interface UpdateDeliveryTimingEventCF { + app_id: string + device_id: string + action: string + version_name: string + metadata: StatsMetadata | null + created_at: string +} + +export interface ReadUpdateDeliveryTimingEventsCFParams { + start_date: string + end_date: string + actions: string[] + /** When set, restrict to these app ids. Omit for platform-wide scans. */ + app_ids?: string[] + limit?: number +} + +export function buildUpdateDeliveryTimingEventsCFQuery(params: ReadUpdateDeliveryTimingEventsCFParams): string { + const limit = normalizeAnalyticsLimit(params.limit, MAX_ANALYTICS_QUERY_LIMIT) + const actionsList = params.actions.map(action => `'${escapeSqlString(action)}'`).join(', ') + const appFilter = params.app_ids?.length + ? ( + params.app_ids.length === 1 + ? `AND index1 = '${escapeSqlString(params.app_ids[0])}'` + : `AND index1 IN (${params.app_ids.map(id => `'${escapeSqlString(id)}'`).join(', ')})` + ) + : '' + + return `SELECT + index1 AS app_id, + blob1 AS device_id, + blob2 AS action, + blob3 AS version_name, + blob4 AS metadata, + timestamp AS created_at +FROM app_log +WHERE + timestamp >= toDateTime('${formatDateCF(params.start_date)}') + AND timestamp < toDateTime('${formatDateCF(params.end_date)}') + AND blob2 IN (${actionsList}) + ${appFilter} +GROUP BY app_id, device_id, action, version_name, metadata, created_at +ORDER BY created_at DESC +LIMIT ${limit}` +} + +export async function readUpdateDeliveryTimingEventsCF( + c: Context, + params: ReadUpdateDeliveryTimingEventsCFParams, +): Promise { + if (!c.env.APP_LOG) + return [] + + if (!params.actions.length) + return [] + + // Empty app_ids array means "no apps in scope" (e.g. empty org), not platform-wide. + if (params.app_ids && params.app_ids.length === 0) + return [] + + const query = buildUpdateDeliveryTimingEventsCFQuery(params) + cloudlog({ requestId: c.get('requestId'), message: 'readUpdateDeliveryTimingEventsCF query', query }) + try { + const rows = await runQueryToCFA<{ + app_id: string + device_id: string + action: string + version_name: string + metadata: string | null + created_at: string + }>(c, query) + return rows.map(row => ({ + app_id: row.app_id, + device_id: row.device_id, + action: row.action, + version_name: row.version_name || 'unknown', + metadata: parseStatsMetadata(row.metadata), + created_at: row.created_at, + })) + } + catch (e) { + cloudlogErr({ requestId: c.get('requestId'), message: 'Error reading update delivery timing events', error: serializeError(e), query }) + throw e + } +} + function buildStatsInsightsActionFilter(actions?: string[]) { if (!actions?.length) return '' diff --git a/tests/helpers/collectAnalyticsEngineSqlFixtures.ts b/tests/helpers/collectAnalyticsEngineSqlFixtures.ts index 48ff343920..532b4d9f90 100644 --- a/tests/helpers/collectAnalyticsEngineSqlFixtures.ts +++ b/tests/helpers/collectAnalyticsEngineSqlFixtures.ts @@ -1,6 +1,7 @@ import type { Context } from 'hono' import { buildReadDevicesCFQuery, + buildUpdateDeliveryTimingEventsCFQuery, countDevicesCF, countInstallSourcesCF, countUpdatesFromLogsCF, @@ -11,27 +12,28 @@ import { getAdminDistributionMetrics, getAdminFailureMetrics, getAdminMauTrend, - getAdminOrgMetrics, getAdminOnboardingTelemetry, + getAdminOrgMetrics, getAdminPlatformOverview, getAdminStorageTrend, getAdminSuccessRate, getAdminSuccessRateTrend, getAdminUploadMetrics, - getPublicLiveUpdateMetricsCF, getPluginBreakdownCF, + getPublicLiveUpdateMetricsCF, getUpdateStatsCF, readActiveAppsCF, readBandwidthUsageCF, + readDevicesCF, readDeviceUsageCF, readDeviceVersionCountsCF, - readDevicesCF, readLastMonthDevicesByPlatformCF, readLastMonthDevicesCF, readLastMonthUpdatesCF, readNativeVersionUsageCF, readStatsCF, readStatsVersionCF, + readUpdateDeliveryTimingEventsCF, } from '../../supabase/functions/_backend/utils/cloudflare.ts' export interface AnalyticsEngineSqlFixture { @@ -215,6 +217,16 @@ export async function collectAnalyticsEngineSqlFixtures(): Promise readUpdateDeliveryTimingEventsCF(context, { + start_date: SAMPLE_START, + end_date: SAMPLE_END, + actions: ['download_complete', 'download_0', 'download_zip_start'], + app_ids: [SAMPLE_APP_ID], + limit: 10, + })) await captureCall('readStatsCF', () => readStatsCF(context, { app_id: SAMPLE_APP_ID, start_date: SAMPLE_START, diff --git a/tests/update-delivery-stats.unit.test.ts b/tests/update-delivery-stats.unit.test.ts index f4916bb0c4..1aff34e3c8 100644 --- a/tests/update-delivery-stats.unit.test.ts +++ b/tests/update-delivery-stats.unit.test.ts @@ -73,4 +73,110 @@ describe('update delivery stats helpers', () => { expect(updateDeliveryStatsTestUtils.toMetric('')).toBeNull() expect(updateDeliveryStatsTestUtils.toMetric(12.4)).toBe(12) }) + + it.concurrent('parses duration metadata strings', () => { + expect(updateDeliveryStatsTestUtils.parseMetaDurationMs({ duration_ms: '1250.5' })).toBe(1250.5) + expect(updateDeliveryStatsTestUtils.parseMetaDurationMs({ duration: '900' })).toBe(900) + expect(updateDeliveryStatsTestUtils.parseMetaDurationMs({ duration_ms: 'nope' })).toBeNull() + expect(updateDeliveryStatsTestUtils.parseMetaDurationMs(null)).toBeNull() + }) + + it('pairs start/complete events when metadata duration is missing', () => { + const periodStartMs = Date.parse('2026-07-02T00:00:00.000Z') + const samples = updateDeliveryStatsTestUtils.buildDeliveriesFromEvents([ + { + app_id: 'com.demo.app', + device_id: 'device-1', + action: 'download_0', + version_name: '1.2.3', + metadata: null, + created_at: '2026-07-02T10:00:00.000Z', + }, + { + app_id: 'com.demo.app', + device_id: 'device-1', + action: 'download_complete', + version_name: '1.2.3', + metadata: null, + created_at: '2026-07-02T10:00:01.500Z', + }, + { + app_id: 'com.demo.app', + device_id: 'device-2', + action: 'download_complete', + version_name: '1.2.3', + metadata: { duration_ms: '2200' }, + created_at: '2026-07-02T11:00:00.000Z', + }, + ], { periodStartMs, allowPairing: true }) + + expect(samples).toEqual([ + { + day: '2026-07-02', + app_id: 'com.demo.app', + device_id: 'device-1', + duration_ms: 1500, + }, + { + day: '2026-07-02', + app_id: 'com.demo.app', + device_id: 'device-2', + duration_ms: 2200, + }, + ]) + }) + + it('skips pairing for platform-style metadata-only mode', () => { + const periodStartMs = Date.parse('2026-07-02T00:00:00.000Z') + const samples = updateDeliveryStatsTestUtils.buildDeliveriesFromEvents([ + { + app_id: 'com.demo.app', + device_id: 'device-1', + action: 'download_0', + version_name: '1.2.3', + metadata: null, + created_at: '2026-07-02T10:00:00.000Z', + }, + { + app_id: 'com.demo.app', + device_id: 'device-1', + action: 'download_complete', + version_name: '1.2.3', + metadata: null, + created_at: '2026-07-02T10:00:01.500Z', + }, + { + app_id: 'com.demo.app', + device_id: 'device-2', + action: 'download_complete', + version_name: '1.2.3', + metadata: { duration_ms: '2200' }, + created_at: '2026-07-02T11:00:00.000Z', + }, + ], { periodStartMs, allowPairing: false }) + + expect(samples).toEqual([ + { + day: '2026-07-02', + app_id: 'com.demo.app', + device_id: 'device-2', + duration_ms: 2200, + }, + ]) + }) + + it('aggregates daily and overview percentiles from samples', () => { + const { dailyRows, overviewRow } = updateDeliveryStatsTestUtils.aggregateDeliverySamples([ + { day: '2026-07-01', app_id: 'a', device_id: 'd1', duration_ms: 100 }, + { day: '2026-07-01', app_id: 'a', device_id: 'd2', duration_ms: 300 }, + { day: '2026-07-02', app_id: 'a', device_id: 'd1', duration_ms: 200 }, + ]) + + expect(dailyRows).toHaveLength(2) + expect(dailyRows[0]).toMatchObject({ day: '2026-07-01', samples: 2, p50_ms: 200 }) + expect(overviewRow.samples).toBe(3) + expect(overviewRow.devices).toBe(2) + expect(overviewRow.p50_ms).toBe(200) + expect(updateDeliveryStatsTestUtils.percentileCont([100, 200, 300], 0.5)).toBe(200) + }) }) From 571dfdad36d3ae5c89aa3bedd73a342ce225895d Mon Sep 17 00:00:00 2001 From: Martin Donadieu Date: Tue, 28 Jul 2026 11:25:17 +0300 Subject: [PATCH 2/2] fix(dashboard): chunk delivery latency CF reads by day Avoid skewing percentiles when APP_LOG returns only the newest 50k rows for a multi-day window on busy apps. Co-authored-by: Cursor --- .../_backend/private/update_delivery_stats.ts | 39 +++++++++++++++++-- .../functions/_backend/utils/cloudflare.ts | 2 +- 2 files changed, 36 insertions(+), 5 deletions(-) diff --git a/supabase/functions/_backend/private/update_delivery_stats.ts b/supabase/functions/_backend/private/update_delivery_stats.ts index 013bba1914..9629188f1c 100644 --- a/supabase/functions/_backend/private/update_delivery_stats.ts +++ b/supabase/functions/_backend/private/update_delivery_stats.ts @@ -544,6 +544,37 @@ async function readUpdateDeliveryStatsSB( } } +async function readUpdateDeliveryTimingEventsCFChunked( + c: Context, + params: { + queryStart: Dayjs + endExclusive: Dayjs + actions: string[] + appIds?: string[] + }, +) { + // AE SQL has a hard row cap and no JOIN. Fetch UTC day windows so busy apps + // keep coverage across the whole period instead of only the newest 50k rows. + const events: UpdateDeliveryTimingEventCF[] = [] + let cursor = params.queryStart.utc().startOf('day') + const end = params.endExclusive.utc() + + while (cursor.isBefore(end)) { + const next = cursor.add(1, 'day') + const chunkEnd = next.isBefore(end) ? next : end + const chunk = await readUpdateDeliveryTimingEventsCF(c, { + start_date: cursor.toISOString(), + end_date: chunkEnd.toISOString(), + actions: params.actions, + app_ids: params.appIds, + }) + events.push(...chunk) + cursor = next + } + + return events +} + async function readUpdateDeliveryStatsCF( c: Context, scope: UpdateDeliveryScope, @@ -571,11 +602,11 @@ async function readUpdateDeliveryStatsCF( appIds = await listOrgAppIds(c, scopeId) } - const events = await readUpdateDeliveryTimingEventsCF(c, { - start_date: queryStart.toISOString(), - end_date: endExclusive.toISOString(), + const events = await readUpdateDeliveryTimingEventsCFChunked(c, { + queryStart, + endExclusive, actions: allowPairing ? [...timingActions] : [...endActions], - app_ids: appIds, + appIds, }) const samples = buildDeliveriesFromEvents(events, { diff --git a/supabase/functions/_backend/utils/cloudflare.ts b/supabase/functions/_backend/utils/cloudflare.ts index 89bcca8364..42f153f235 100644 --- a/supabase/functions/_backend/utils/cloudflare.ts +++ b/supabase/functions/_backend/utils/cloudflare.ts @@ -1283,7 +1283,7 @@ WHERE AND blob2 IN (${actionsList}) ${appFilter} GROUP BY app_id, device_id, action, version_name, metadata, created_at -ORDER BY created_at DESC +ORDER BY created_at ASC LIMIT ${limit}` }