fix(dashboard): read update delivery latency from Analytics Engine

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 @@
   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<string>(endActions)
+const startActionSet = new Set<string>(startActions)
 
 const durationExpression = String.raw`CASE
   WHEN s.metadata ? 'duration_ms'
@@ -238,6 +252,158 @@
   return Math.round(numeric * factor) / factor
 }
 
+function parseMetaDurationMs(metadata: Record<string, string> | 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<string, number[]>()
+  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<string, number[]>()
+  const allDurations: number[] = []
+  const devices = new Set<string>()
+
+  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 @@
     throw simpleError('not_admin', 'Not admin - only admin users can access platform delivery latency')
 }
 
-async function readUpdateDeliveryStats(
+async function listOrgAppIds(c: Context<MiddlewareKeyVariables>, orgId: string): Promise<string[]> {
+  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<MiddlewareKeyVariables>,
   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,14 +536,133 @@
     })
   }
   catch (error) {
-    logPgError(c, 'readUpdateDeliveryStats', error)
+    logPgError(c, 'readUpdateDeliveryStatsSB', error)
     throw error
   }
   finally {
     await closeClient(c, db)
   }
 }
 
+async function readUpdateDeliveryTimingEventsCFChunked(
+  c: Context<MiddlewareKeyVariables>,
+  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<MiddlewareKeyVariables>,
+  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 readUpdateDeliveryTimingEventsCFChunked(c, {
+    queryStart,
+    endExclusive,
+    actions: allowPairing ? [...timingActions] : [...endActions],
+    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<MiddlewareKeyVariables>,
+  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<MiddlewareKeyVariables>()
 
 app.use('/', useCors)
@@ -423,8 +720,12 @@
 
 export const updateDeliveryStatsTestUtils = {
   buildUpdateDeliveryResponse,
+  buildDeliveriesFromEvents,
+  aggregateDeliverySamples,
   generateDateLabels,
   normalizePeriodDays,
   normalizeScope,
+  parseMetaDurationMs,
+  percentileCont,
   toMetric,
 }