diff --git a/cloudflare_workers/api/index.ts b/cloudflare_workers/api/index.ts index e04f7391aa..f79a5117e4 100644 --- a/cloudflare_workers/api/index.ts +++ b/cloudflare_workers/api/index.ts @@ -5,8 +5,6 @@ import { app as admin_credits } from '../../supabase/functions/_backend/private/ import { app as admin_stats } from '../../supabase/functions/_backend/private/admin_stats.ts' import { app as channel_device } from '../../supabase/functions/_backend/private/channel_device.ts' import { app as channel_stats } from '../../supabase/functions/_backend/private/channel_stats.ts' -import { app as native_observe_stats } from '../../supabase/functions/_backend/private/native_observe_stats.ts' -import { app as update_delivery_stats } from '../../supabase/functions/_backend/private/update_delivery_stats.ts' import { app as config } from '../../supabase/functions/_backend/private/config.ts' import { app as configBuilder } from '../../supabase/functions/_backend/private/config_builder.ts' import { app as create_device } from '../../supabase/functions/_backend/private/create_device.ts' @@ -19,6 +17,7 @@ import { app as invite_existing_user_to_org } from '../../supabase/functions/_ba import { app as invite_new_user_to_org } from '../../supabase/functions/_backend/private/invite_new_user_to_org.ts' import { app as latency } from '../../supabase/functions/_backend/private/latency.ts' import { app as log_as } from '../../supabase/functions/_backend/private/log_as.ts' +import { app as native_observe_stats } from '../../supabase/functions/_backend/private/native_observe_stats.ts' import { app as plans } from '../../supabase/functions/_backend/private/plans.ts' import { app as publicStats } from '../../supabase/functions/_backend/private/public_stats.ts' import { app as replay } from '../../supabase/functions/_backend/private/replay.ts' @@ -35,6 +34,7 @@ import { app as stats_priv } from '../../supabase/functions/_backend/private/sta import { app as storeTop } from '../../supabase/functions/_backend/private/store_top.ts' import { app as stripe_checkout } from '../../supabase/functions/_backend/private/stripe_checkout.ts' import { app as stripe_portal } from '../../supabase/functions/_backend/private/stripe_portal.ts' +import { app as update_delivery_stats } from '../../supabase/functions/_backend/private/update_delivery_stats.ts' import { app as validate_password_compliance } from '../../supabase/functions/_backend/private/validate_password_compliance.ts' import { app as verify_email_otp } from '../../supabase/functions/_backend/private/verify_email_otp.ts' import { app as apikey } from '../../supabase/functions/_backend/public/apikey/index.ts' @@ -48,6 +48,7 @@ import { app as notifications } from '../../supabase/functions/_backend/public/n import { app as ok } from '../../supabase/functions/_backend/public/ok.ts' import { app as organization } from '../../supabase/functions/_backend/public/organization/index.ts' import { app as pluginRegions } from '../../supabase/functions/_backend/public/plugin_regions.ts' +import { app as queue_health } from '../../supabase/functions/_backend/public/queue_health.ts' import { app as replication } from '../../supabase/functions/_backend/public/replication.ts' import { app as statistics } from '../../supabase/functions/_backend/public/statistics/index.ts' import { app as translation } from '../../supabase/functions/_backend/public/translation.ts' @@ -104,6 +105,7 @@ app.route('/webhooks', webhooks) app.route('/app', appEndpoint) app.route('/build', build) app.route('/replication', replication) +app.route('/queue_health', queue_health) app.route('/check_cpu_usage', check_cpu_usage) app.route('/translation', translation) app.route('/plugin_regions', pluginRegions) diff --git a/supabase/config.toml b/supabase/config.toml index 6944eef034..eeb4478500 100644 --- a/supabase/config.toml +++ b/supabase/config.toml @@ -440,3 +440,9 @@ import_map = "./functions/deno.json" [functions.check_cpu_usage] verify_jwt = false import_map = "./functions/deno.json" + +# Queue Health + +[functions.queue_health] +verify_jwt = false +import_map = "./functions/deno.json" diff --git a/supabase/functions/_backend/public/queue_health.ts b/supabase/functions/_backend/public/queue_health.ts new file mode 100644 index 0000000000..514c4f6548 --- /dev/null +++ b/supabase/functions/_backend/public/queue_health.ts @@ -0,0 +1,535 @@ +import { honoFactory, useCors } from '../utils/hono.ts' +import { cloudlogErr } from '../utils/logging.ts' +import { closeClient, getPgClient, logPgError } from '../utils/pg.ts' +import { validatePlatformAdminOrApiSecret } from '../utils/platform_admin_access.ts' + +type QueueStatus = 'ok' | 'ko' +type SqlCountValue = string | number | null + +export const STUCK_READ_CT_THRESHOLD = 5 +// cleanup_queue_messages purges archives older than 2 days; allow 1 day of batch lag. +export const ARCHIVE_STALE_SECONDS = 3 * 24 * 60 * 60 +export const ARCHIVE_RECENT_WINDOW_SECONDS = 60 * 60 +export const DEFAULT_ARCHIVE_RECENT_THRESHOLD = 5_000 +export const DEFAULT_QUEUE_DEPTH_THRESHOLD = 50_000 +export const DEFAULT_NEVER_READ_STALE_SECONDS = 60 * 60 +export const MIN_NEVER_READ_STALE_SECONDS = 5 * 60 +export const NEVER_READ_INTERVAL_MULTIPLIER = 3 + +const SAFE_QUEUE_NAME = /^[a-z_]\w*$/i + +export interface QueueHealthThresholds { + stuck_read_ct: number + archive_stale_seconds: number + archive_recent_window_seconds: number + archive_recent_threshold: number + queue_depth_threshold: number + default_never_read_stale_seconds: number + min_never_read_stale_seconds: number + never_read_interval_multiplier: number +} + +export interface QueueMetrics { + queue_name: string + queue_table_exists: boolean + archive_table_exists: boolean + queue_count: number + never_read_count: number + never_read_stale_count: number + stuck_count: number + max_read_ct: number | null + oldest_message_age_seconds: number | null + archive_count: number + archive_stale_count: number + archive_recent_count: number + oldest_archive_age_seconds: number | null + expected_interval_seconds: number | null + never_read_stale_seconds: number +} + +export interface QueueHealthResult { + queue_name: string + status: QueueStatus + reasons: string[] + reason_details: Record + queue_table_exists: boolean + archive_table_exists: boolean + queue_count: number + never_read_count: number + never_read_stale_count: number + stuck_count: number + max_read_ct: number | null + oldest_message_age_seconds: number | null + archive_count: number + archive_stale_count: number + archive_recent_count: number + oldest_archive_age_seconds: number | null + expected_interval_seconds: number | null + never_read_stale_seconds: number +} + +function toNumber(value: unknown): number | null { + if (value === null || value === undefined) + return null + const num = Number(value) + if (!Number.isFinite(num)) + return null + return num +} + +export function isSafeQueueName(queueName: string): boolean { + return SAFE_QUEUE_NAME.test(queueName) +} + +export function cronTaskIntervalSeconds(task: { + second_interval?: number | null + minute_interval?: number | null + hour_interval?: number | null + run_at_hour?: number | null + run_on_dow?: number | null + run_on_day?: number | null +}): number | null { + if (task.second_interval != null && task.second_interval > 0) + return task.second_interval + if (task.minute_interval != null && task.minute_interval > 0) + return task.minute_interval * 60 + if (task.hour_interval != null && task.hour_interval > 0) + return task.hour_interval * 3600 + // Calendar schedules: prefer the coarsest cadence marker first. + if (task.run_on_day != null) + return 31 * 24 * 60 * 60 + if (task.run_on_dow != null) + return 7 * 24 * 60 * 60 + if (task.run_at_hour != null) + return 24 * 60 * 60 + return null +} + +export function parseCronQueueTargets(target: unknown): string[] { + if (typeof target === 'string') { + const trimmed = target.trim() + if (!trimmed) + return [] + if (trimmed.startsWith('[')) { + try { + return parseCronQueueTargets(JSON.parse(trimmed)) + } + catch { + return isSafeQueueName(trimmed) ? [trimmed] : [] + } + } + return isSafeQueueName(trimmed) ? [trimmed] : [] + } + + if (Array.isArray(target)) { + return target + .filter((value): value is string => typeof value === 'string' && isSafeQueueName(value)) + } + + return [] +} + +export function buildQueueIntervalMap( + schedules: Array<{ + task_type: string + target: unknown + second_interval?: number | null + minute_interval?: number | null + hour_interval?: number | null + run_at_hour?: number | null + run_on_dow?: number | null + run_on_day?: number | null + }>, +): Map { + const intervals = new Map() + + for (const schedule of schedules) { + if (schedule.task_type !== 'function_queue' && schedule.task_type !== 'queue') + continue + + const interval = cronTaskIntervalSeconds(schedule) + if (interval == null) + continue + + for (const queueName of parseCronQueueTargets(schedule.target)) { + const existing = intervals.get(queueName) + if (existing == null || interval < existing) + intervals.set(queueName, interval) + } + } + + return intervals +} + +export function resolveNeverReadStaleSeconds( + expectedIntervalSeconds: number | null, + thresholds: Pick, +): number { + if (expectedIntervalSeconds == null || expectedIntervalSeconds <= 0) + return thresholds.default_never_read_stale_seconds + + return Math.max( + expectedIntervalSeconds * thresholds.never_read_interval_multiplier, + thresholds.min_never_read_stale_seconds, + ) +} + +export function evaluateQueueHealth( + metrics: QueueMetrics, + thresholds: QueueHealthThresholds, +): QueueHealthResult { + const reasons: string[] = [] + const reason_details: Record = {} + + if (!metrics.queue_table_exists) { + reasons.push('missing_queue_table') + reason_details.missing_queue_table = `Queue table pgmq.q_${metrics.queue_name} is missing while the queue is registered in pgmq.meta` + } + + if (!metrics.archive_table_exists) { + reasons.push('missing_archive_table') + reason_details.missing_archive_table = `Archive table pgmq.a_${metrics.queue_name} is missing while the queue is registered in pgmq.meta` + } + + if (metrics.queue_table_exists && metrics.never_read_stale_count > 0) { + reasons.push('never_read_stale') + reason_details.never_read_stale = `${metrics.never_read_stale_count} message(s) still have read_ct=0 after ${metrics.never_read_stale_seconds}s (oldest age ${metrics.oldest_message_age_seconds ?? 'unknown'}s). Consumers are not reading this queue in time.` + } + + if (metrics.queue_table_exists && metrics.stuck_count > 0) { + reasons.push('stuck_high_read_ct') + reason_details.stuck_high_read_ct = `${metrics.stuck_count} message(s) have read_ct > ${thresholds.stuck_read_ct} (max ${metrics.max_read_ct ?? 'unknown'}). Messages are being retried without successful ack/delete.` + } + + if (metrics.queue_table_exists && metrics.queue_count > thresholds.queue_depth_threshold) { + reasons.push('queue_depth_exceeded') + reason_details.queue_depth_exceeded = `Queue depth is ${metrics.queue_count}, above threshold ${thresholds.queue_depth_threshold}. Consumers cannot keep up.` + } + + if (metrics.archive_table_exists && metrics.archive_stale_count > 0) { + reasons.push('archive_stale') + reason_details.archive_stale = `${metrics.archive_stale_count} archived message(s) are older than ${thresholds.archive_stale_seconds}s. cleanup_queue_messages is not keeping archives from ramping up.` + } + + if (metrics.archive_table_exists && metrics.archive_recent_count > thresholds.archive_recent_threshold) { + reasons.push('archive_ramping') + reason_details.archive_ramping = `${metrics.archive_recent_count} message(s) were archived in the last ${thresholds.archive_recent_window_seconds}s, above threshold ${thresholds.archive_recent_threshold}. Archive growth indicates consumers are failing and dumping work into archive.` + } + + return { + queue_name: metrics.queue_name, + status: reasons.length > 0 ? 'ko' : 'ok', + reasons, + reason_details, + queue_table_exists: metrics.queue_table_exists, + archive_table_exists: metrics.archive_table_exists, + queue_count: metrics.queue_count, + never_read_count: metrics.never_read_count, + never_read_stale_count: metrics.never_read_stale_count, + stuck_count: metrics.stuck_count, + max_read_ct: metrics.max_read_ct, + oldest_message_age_seconds: metrics.oldest_message_age_seconds, + archive_count: metrics.archive_count, + archive_stale_count: metrics.archive_stale_count, + archive_recent_count: metrics.archive_recent_count, + oldest_archive_age_seconds: metrics.oldest_archive_age_seconds, + expected_interval_seconds: metrics.expected_interval_seconds, + never_read_stale_seconds: metrics.never_read_stale_seconds, + } +} + +export function buildQueueHealthCriteria(thresholds: QueueHealthThresholds) { + return { + never_read_stale: { + healthy_when: 'No queue messages remain with read_ct=0 longer than the queue stale threshold (derived from cron interval when known).', + unhealthy_when: 'Messages sit unread long enough that consumers are likely stuck or not scheduled.', + default_threshold_seconds: thresholds.default_never_read_stale_seconds, + min_threshold_seconds: thresholds.min_never_read_stale_seconds, + interval_multiplier: thresholds.never_read_interval_multiplier, + }, + stuck_high_read_ct: { + healthy_when: `No queue messages have read_ct > ${thresholds.stuck_read_ct}.`, + unhealthy_when: 'Messages keep retrying past the stuck threshold without successful processing.', + threshold: thresholds.stuck_read_ct, + }, + queue_depth_exceeded: { + healthy_when: `Queue depth stays at or below ${thresholds.queue_depth_threshold}.`, + unhealthy_when: 'Backlog is too large for consumers to drain safely.', + threshold: thresholds.queue_depth_threshold, + }, + archive_stale: { + healthy_when: `No archived rows older than ${thresholds.archive_stale_seconds}s (cleanup window).`, + unhealthy_when: 'Old archive rows remain, so archive storage is ramping instead of being cleaned.', + threshold_seconds: thresholds.archive_stale_seconds, + }, + archive_ramping: { + healthy_when: `Archived rows created in the last ${thresholds.archive_recent_window_seconds}s stay at or below ${thresholds.archive_recent_threshold}.`, + unhealthy_when: 'Recent archive volume is abnormally high, usually from retry-budget exhaustion.', + window_seconds: thresholds.archive_recent_window_seconds, + threshold: thresholds.archive_recent_threshold, + }, + missing_queue_table: { + healthy_when: 'Every registered queue has pgmq.q_.', + unhealthy_when: 'pgmq.meta references a queue whose queue table is missing.', + }, + missing_archive_table: { + healthy_when: 'Every registered queue has pgmq.a_.', + unhealthy_when: 'pgmq.meta references a queue whose archive table is missing.', + }, + } +} + +function defaultThresholds(): QueueHealthThresholds { + return { + stuck_read_ct: STUCK_READ_CT_THRESHOLD, + archive_stale_seconds: ARCHIVE_STALE_SECONDS, + archive_recent_window_seconds: ARCHIVE_RECENT_WINDOW_SECONDS, + archive_recent_threshold: DEFAULT_ARCHIVE_RECENT_THRESHOLD, + queue_depth_threshold: DEFAULT_QUEUE_DEPTH_THRESHOLD, + default_never_read_stale_seconds: DEFAULT_NEVER_READ_STALE_SECONDS, + min_never_read_stale_seconds: MIN_NEVER_READ_STALE_SECONDS, + never_read_interval_multiplier: NEVER_READ_INTERVAL_MULTIPLIER, + } +} + +async function listQueues(client: ReturnType): Promise { + const { rows } = await client.query<{ queue_name: string }>( + 'SELECT queue_name FROM pgmq.list_queues() ORDER BY queue_name', + ) + return rows + .map(row => row.queue_name) + .filter((name): name is string => typeof name === 'string' && isSafeQueueName(name)) +} + +async function loadQueueIntervals(client: ReturnType): Promise> { + const { rows } = await client.query<{ + task_type: string + target: unknown + second_interval: number | null + minute_interval: number | null + hour_interval: number | null + run_at_hour: number | null + run_on_dow: number | null + run_on_day: number | null + }>(` + SELECT task_type::text AS task_type, + target, + second_interval, + minute_interval, + hour_interval, + run_at_hour, + run_on_dow, + run_on_day + FROM public.cron_tasks + WHERE enabled = true + AND task_type IN ('function_queue', 'queue') + `) + + return buildQueueIntervalMap(rows) +} + +async function fetchQueueMetrics( + client: ReturnType, + queueName: string, + expectedIntervalSeconds: number | null, + thresholds: QueueHealthThresholds, +): Promise { + if (!isSafeQueueName(queueName)) + throw new Error(`Invalid queue name: ${queueName}`) + + const neverReadStaleSeconds = resolveNeverReadStaleSeconds(expectedIntervalSeconds, thresholds) + + const { rows: existenceRows } = await client.query<{ + queue_table_exists: boolean + archive_table_exists: boolean + }>( + ` + SELECT + to_regclass($1) IS NOT NULL AS queue_table_exists, + to_regclass($2) IS NOT NULL AS archive_table_exists + `, + [`pgmq.q_${queueName}`, `pgmq.a_${queueName}`], + ) + + const queueExists = Boolean(existenceRows[0]?.queue_table_exists) + const archiveExists = Boolean(existenceRows[0]?.archive_table_exists) + + let queue_count = 0 + let never_read_count = 0 + let never_read_stale_count = 0 + let stuck_count = 0 + let max_read_ct: number | null = null + let oldest_message_age_seconds: number | null = null + + if (queueExists) { + const { rows } = await client.query<{ + queue_count: SqlCountValue + never_read_count: SqlCountValue + never_read_stale_count: SqlCountValue + stuck_count: SqlCountValue + max_read_ct: SqlCountValue + oldest_message_age_seconds: SqlCountValue + }>( + ` + SELECT + COUNT(*)::bigint AS queue_count, + COUNT(*) FILTER (WHERE read_ct = 0)::bigint AS never_read_count, + COUNT(*) FILTER ( + WHERE read_ct = 0 + AND enqueued_at < now() - ($1::text || ' seconds')::interval + )::bigint AS never_read_stale_count, + COUNT(*) FILTER (WHERE read_ct > $2)::bigint AS stuck_count, + MAX(read_ct)::bigint AS max_read_ct, + EXTRACT(EPOCH FROM (now() - MIN(enqueued_at)))::numeric AS oldest_message_age_seconds + FROM pgmq.q_${queueName} + `, + [String(neverReadStaleSeconds), thresholds.stuck_read_ct], + ) + + queue_count = toNumber(rows[0]?.queue_count) ?? 0 + never_read_count = toNumber(rows[0]?.never_read_count) ?? 0 + never_read_stale_count = toNumber(rows[0]?.never_read_stale_count) ?? 0 + stuck_count = toNumber(rows[0]?.stuck_count) ?? 0 + max_read_ct = toNumber(rows[0]?.max_read_ct ?? null) + oldest_message_age_seconds = toNumber(rows[0]?.oldest_message_age_seconds ?? null) + } + + let archive_count = 0 + let archive_stale_count = 0 + let archive_recent_count = 0 + let oldest_archive_age_seconds: number | null = null + + if (archiveExists) { + const { rows } = await client.query<{ + archive_count: SqlCountValue + archive_stale_count: SqlCountValue + archive_recent_count: SqlCountValue + oldest_archive_age_seconds: SqlCountValue + }>( + ` + SELECT + COUNT(*)::bigint AS archive_count, + COUNT(*) FILTER ( + WHERE archived_at < now() - ($1::text || ' seconds')::interval + )::bigint AS archive_stale_count, + COUNT(*) FILTER ( + WHERE archived_at >= now() - ($2::text || ' seconds')::interval + )::bigint AS archive_recent_count, + EXTRACT(EPOCH FROM (now() - MIN(archived_at)))::numeric AS oldest_archive_age_seconds + FROM pgmq.a_${queueName} + `, + [String(thresholds.archive_stale_seconds), String(thresholds.archive_recent_window_seconds)], + ) + + archive_count = toNumber(rows[0]?.archive_count) ?? 0 + archive_stale_count = toNumber(rows[0]?.archive_stale_count) ?? 0 + archive_recent_count = toNumber(rows[0]?.archive_recent_count) ?? 0 + oldest_archive_age_seconds = toNumber(rows[0]?.oldest_archive_age_seconds ?? null) + } + + return { + queue_name: queueName, + queue_table_exists: queueExists, + archive_table_exists: archiveExists, + queue_count, + never_read_count, + never_read_stale_count, + stuck_count, + max_read_ct, + oldest_message_age_seconds, + archive_count, + archive_stale_count, + archive_recent_count, + oldest_archive_age_seconds, + expected_interval_seconds: expectedIntervalSeconds, + never_read_stale_seconds: neverReadStaleSeconds, + } +} + +export const app = honoFactory.createApp() + +app.use('*', useCors) + +app.get('/', async (c) => { + await validatePlatformAdminOrApiSecret(c, { + logPrefix: 'queue_health', + forbiddenMessage: 'Not admin - only admin users can access queue health', + }) + + const thresholds = defaultThresholds() + const pgClient = getPgClient(c, false) + + try { + const queueNames = await listQueues(pgClient) + const intervalMap = await loadQueueIntervals(pgClient) + + const metricsList = await Promise.all( + queueNames.map(queueName => fetchQueueMetrics( + pgClient, + queueName, + intervalMap.get(queueName) ?? null, + thresholds, + )), + ) + + const queues = metricsList.map(metrics => evaluateQueueHealth(metrics, thresholds)) + const unhealthy = queues.filter(queue => queue.status === 'ko') + // Empty registry is not a processing failure (fresh installs / no pgmq queues yet). + const overallStatus: QueueStatus = unhealthy.length > 0 ? 'ko' : 'ok' + + const maxQueueDepth = queues.reduce((max, queue) => Math.max(max, queue.queue_count), 0) + const maxArchiveRecent = queues.reduce((max, queue) => Math.max(max, queue.archive_recent_count), 0) + const totalStuck = queues.reduce((sum, queue) => sum + queue.stuck_count, 0) + const totalNeverReadStale = queues.reduce((sum, queue) => sum + queue.never_read_stale_count, 0) + const totalArchiveStale = queues.reduce((sum, queue) => sum + queue.archive_stale_count, 0) + + return c.json({ + status: overallStatus, + checked_at: new Date().toISOString(), + no_queues_registered: queues.length === 0, + queue_count: queues.length, + healthy_count: queues.length - unhealthy.length, + unhealthy_count: unhealthy.length, + max_queue_depth: maxQueueDepth, + max_archive_recent_count: maxArchiveRecent, + total_stuck_count: totalStuck, + total_never_read_stale_count: totalNeverReadStale, + total_archive_stale_count: totalArchiveStale, + thresholds, + criteria: buildQueueHealthCriteria(thresholds), + unhealthy_queues: unhealthy.map(queue => ({ + queue_name: queue.queue_name, + status: queue.status, + reasons: queue.reasons, + reason_details: queue.reason_details, + })), + queues, + }, overallStatus === 'ok' ? 200 : 503) + } + catch (error) { + logPgError(c, 'queue_health', error) + cloudlogErr({ requestId: c.get('requestId'), message: 'queue_health_error', error }) + return c.json({ + status: 'ko', + error: 'queue_health_error', + message: 'Failed to check queue health', + checked_at: new Date().toISOString(), + no_queues_registered: null, + queue_count: 0, + healthy_count: 0, + unhealthy_count: 0, + max_queue_depth: 0, + max_archive_recent_count: 0, + total_stuck_count: 0, + total_never_read_stale_count: 0, + total_archive_stale_count: 0, + thresholds, + criteria: buildQueueHealthCriteria(thresholds), + unhealthy_queues: [], + queues: [], + }, 500) + } + finally { + await closeClient(c, pgClient) + } +}) diff --git a/supabase/functions/_backend/public/replication.ts b/supabase/functions/_backend/public/replication.ts index ed1d2e02ea..8aa9bc8085 100644 --- a/supabase/functions/_backend/public/replication.ts +++ b/supabase/functions/_backend/public/replication.ts @@ -1,11 +1,8 @@ -import type { Context } from 'hono' -import type { MiddlewareKeyVariables } from '../utils/hono.ts' import { sql } from 'drizzle-orm' -import { honoFactory, middlewareAPISecret, quickError, useCors } from '../utils/hono.ts' -import { getClaimsFromJWT } from '../utils/hono_jwt.ts' +import { honoFactory, useCors } from '../utils/hono.ts' import { cloudlogErr } from '../utils/logging.ts' import { closeClient, getDrizzleClient, getPgClient, logPgError } from '../utils/pg.ts' -import { supabaseClient } from '../utils/supabase.ts' +import { validatePlatformAdminOrApiSecret } from '../utils/platform_admin_access.ts' const DEFAULT_THRESHOLD_SECONDS = 180 const DEFAULT_THRESHOLD_BYTES = 16 * 1024 * 1024 @@ -146,50 +143,11 @@ export const app = honoFactory.createApp() app.use('*', useCors) -type ReplicationContext = Context - -async function validateReplicationAccess(c: ReplicationContext) { - const apiSecret = c.req.header('apisecret') - - if (apiSecret) { - await middlewareAPISecret(c, async () => {}) - return - } - - const authorization = c.req.header('authorization') - if (!authorization) { - throw quickError(401, 'no_authorization', 'Authorization header or apisecret is required') - } - - const claims = await getClaimsFromJWT(c, authorization) - if (!claims?.sub) { - cloudlogErr({ requestId: c.get('requestId'), message: 'replication_invalid_jwt' }) - throw quickError(401, 'invalid_jwt', 'Invalid JWT') - } - - c.set('authorization', authorization) - c.set('auth', { - userId: claims.sub, - authType: 'jwt', - apikey: null, - jwt: authorization, - }) - - const userClient = supabaseClient(c, authorization) - const { data: isAdmin, error: adminError } = await userClient.rpc('is_platform_admin') - if (adminError) { - cloudlogErr({ requestId: c.get('requestId'), message: 'replication_is_admin_error', error: adminError }) - throw quickError(500, 'is_admin_error', 'Unable to verify admin rights') - } - - if (!isAdmin) { - cloudlogErr({ requestId: c.get('requestId'), message: 'replication_not_admin', userId: claims.sub }) - throw quickError(403, 'not_admin', 'Not admin - only admin users can access replication status') - } -} - app.get('/', async (c) => { - await validateReplicationAccess(c) + await validatePlatformAdminOrApiSecret(c, { + logPrefix: 'replication', + forbiddenMessage: 'Not admin - only admin users can access replication status', + }) const thresholdSeconds = DEFAULT_THRESHOLD_SECONDS const thresholdBytes = DEFAULT_THRESHOLD_BYTES diff --git a/supabase/functions/_backend/utils/platform_admin_access.ts b/supabase/functions/_backend/utils/platform_admin_access.ts new file mode 100644 index 0000000000..531f613e6b --- /dev/null +++ b/supabase/functions/_backend/utils/platform_admin_access.ts @@ -0,0 +1,54 @@ +import type { Context } from 'hono' +import type { MiddlewareKeyVariables } from './hono.ts' +import { middlewareAPISecret, quickError } from './hono.ts' +import { getClaimsFromJWT } from './hono_jwt.ts' +import { cloudlogErr } from './logging.ts' +import { supabaseClient } from './supabase.ts' + +type PlatformAdminContext = Context + +export async function validatePlatformAdminOrApiSecret( + c: PlatformAdminContext, + options: { + logPrefix: string + forbiddenMessage: string + }, +) { + const apiSecret = c.req.header('apisecret') + + if (apiSecret) { + await middlewareAPISecret(c, async () => {}) + return + } + + const authorization = c.req.header('authorization') + if (!authorization) { + throw quickError(401, 'no_authorization', 'Authorization header or apisecret is required') + } + + const claims = await getClaimsFromJWT(c, authorization) + if (!claims?.sub) { + cloudlogErr({ requestId: c.get('requestId'), message: `${options.logPrefix}_invalid_jwt` }) + throw quickError(401, 'invalid_jwt', 'Invalid JWT') + } + + c.set('authorization', authorization) + c.set('auth', { + userId: claims.sub, + authType: 'jwt', + apikey: null, + jwt: authorization, + }) + + const userClient = supabaseClient(c, authorization) + const { data: isAdmin, error: adminError } = await userClient.rpc('is_platform_admin') + if (adminError) { + cloudlogErr({ requestId: c.get('requestId'), message: `${options.logPrefix}_is_admin_error`, error: adminError }) + throw quickError(500, 'is_admin_error', 'Unable to verify admin rights') + } + + if (!isAdmin) { + cloudlogErr({ requestId: c.get('requestId'), message: `${options.logPrefix}_not_admin`, userId: claims.sub }) + throw quickError(403, 'not_admin', options.forbiddenMessage) + } +} diff --git a/supabase/functions/queue_health/index.ts b/supabase/functions/queue_health/index.ts new file mode 100644 index 0000000000..c159f5b6db --- /dev/null +++ b/supabase/functions/queue_health/index.ts @@ -0,0 +1,10 @@ +import { app } from '../_backend/public/queue_health.ts' +import { createAllCatch, createHono } from '../_backend/utils/hono.ts' +import { version } from '../_backend/utils/version.ts' + +const functionName = 'queue_health' +const appGlobal = createHono(functionName, version) + +appGlobal.route('/', app) +createAllCatch(appGlobal, functionName) +Deno.serve(appGlobal.fetch) diff --git a/tests/queue-health.unit.test.ts b/tests/queue-health.unit.test.ts new file mode 100644 index 0000000000..4c729e3ce0 --- /dev/null +++ b/tests/queue-health.unit.test.ts @@ -0,0 +1,177 @@ +import type { QueueHealthThresholds, QueueMetrics } from '../supabase/functions/_backend/public/queue_health.ts' +import { describe, expect, it } from 'vitest' +import { + ARCHIVE_RECENT_WINDOW_SECONDS, + ARCHIVE_STALE_SECONDS, + buildQueueHealthCriteria, + buildQueueIntervalMap, + cronTaskIntervalSeconds, + DEFAULT_ARCHIVE_RECENT_THRESHOLD, + DEFAULT_NEVER_READ_STALE_SECONDS, + DEFAULT_QUEUE_DEPTH_THRESHOLD, + evaluateQueueHealth, + isSafeQueueName, + MIN_NEVER_READ_STALE_SECONDS, + NEVER_READ_INTERVAL_MULTIPLIER, + parseCronQueueTargets, + resolveNeverReadStaleSeconds, + STUCK_READ_CT_THRESHOLD, +} from '../supabase/functions/_backend/public/queue_health.ts' + +const thresholds: QueueHealthThresholds = { + stuck_read_ct: STUCK_READ_CT_THRESHOLD, + archive_stale_seconds: ARCHIVE_STALE_SECONDS, + archive_recent_window_seconds: ARCHIVE_RECENT_WINDOW_SECONDS, + archive_recent_threshold: DEFAULT_ARCHIVE_RECENT_THRESHOLD, + queue_depth_threshold: DEFAULT_QUEUE_DEPTH_THRESHOLD, + default_never_read_stale_seconds: DEFAULT_NEVER_READ_STALE_SECONDS, + min_never_read_stale_seconds: MIN_NEVER_READ_STALE_SECONDS, + never_read_interval_multiplier: NEVER_READ_INTERVAL_MULTIPLIER, +} + +function createMetrics(overrides: Partial = {}): QueueMetrics { + return { + queue_name: 'on_version_update', + queue_table_exists: true, + archive_table_exists: true, + queue_count: 0, + never_read_count: 0, + never_read_stale_count: 0, + stuck_count: 0, + max_read_ct: null, + oldest_message_age_seconds: null, + archive_count: 0, + archive_stale_count: 0, + archive_recent_count: 0, + oldest_archive_age_seconds: null, + expected_interval_seconds: 10, + never_read_stale_seconds: 300, + ...overrides, + } +} + +describe('queue_health helpers', () => { + it.concurrent('accepts only safe pgmq queue names', () => { + expect(isSafeQueueName('on_version_update')).toBe(true) + expect(isSafeQueueName('q;drop table')).toBe(false) + expect(isSafeQueueName('a-b')).toBe(false) + }) + + it.concurrent('parses cron queue targets from arrays and json strings', () => { + expect(parseCronQueueTargets(['webhook_delivery', 'bad-name', 1])).toEqual(['webhook_delivery']) + expect(parseCronQueueTargets('["on_app_create","on_app_delete"]')).toEqual(['on_app_create', 'on_app_delete']) + expect(parseCronQueueTargets('cron_email')).toEqual(['cron_email']) + }) + + it.concurrent('derives cron intervals and picks the fastest schedule per queue', () => { + expect(cronTaskIntervalSeconds({ second_interval: 10 })).toBe(10) + expect(cronTaskIntervalSeconds({ minute_interval: 5 })).toBe(300) + expect(cronTaskIntervalSeconds({ hour_interval: 2 })).toBe(7200) + expect(cronTaskIntervalSeconds({ run_at_hour: 3 })).toBe(86400) + expect(cronTaskIntervalSeconds({ run_at_hour: 12, run_on_dow: 6 })).toBe(7 * 24 * 60 * 60) + expect(cronTaskIntervalSeconds({ run_at_hour: 12, run_on_day: 1 })).toBe(31 * 24 * 60 * 60) + + const intervals = buildQueueIntervalMap([ + { + task_type: 'function_queue', + target: '["on_version_update"]', + second_interval: 10, + }, + { + task_type: 'function_queue', + target: ['on_version_update', 'admin_stats'], + hour_interval: 2, + }, + { + task_type: 'function', + target: '["ignored"]', + second_interval: 10, + }, + ]) + + expect(intervals.get('on_version_update')).toBe(10) + expect(intervals.get('admin_stats')).toBe(7200) + expect(intervals.has('ignored')).toBe(false) + }) + + it.concurrent('resolves never-read stale threshold from cron interval with a floor', () => { + expect(resolveNeverReadStaleSeconds(null, thresholds)).toBe(DEFAULT_NEVER_READ_STALE_SECONDS) + expect(resolveNeverReadStaleSeconds(10, thresholds)).toBe(MIN_NEVER_READ_STALE_SECONDS) + expect(resolveNeverReadStaleSeconds(7200, thresholds)).toBe(21600) + }) +}) + +describe('evaluateQueueHealth', () => { + it.concurrent('marks a clean queue healthy', () => { + const result = evaluateQueueHealth(createMetrics(), thresholds) + expect(result.status).toBe('ok') + expect(result.reasons).toEqual([]) + }) + + it.concurrent('fails when unread messages sit too long', () => { + const result = evaluateQueueHealth(createMetrics({ + never_read_stale_count: 3, + never_read_count: 3, + oldest_message_age_seconds: 900, + }), thresholds) + + expect(result.status).toBe('ko') + expect(result.reasons).toContain('never_read_stale') + expect(result.reason_details.never_read_stale).toContain('read_ct=0') + }) + + it.concurrent('fails when messages are stuck with high read_ct', () => { + const result = evaluateQueueHealth(createMetrics({ + stuck_count: 2, + max_read_ct: 12, + }), thresholds) + + expect(result.status).toBe('ko') + expect(result.reasons).toContain('stuck_high_read_ct') + expect(result.reason_details.stuck_high_read_ct).toContain('read_ct > 5') + }) + + it.concurrent('fails when archive cleanup is lagging or archive is ramping', () => { + const stale = evaluateQueueHealth(createMetrics({ + archive_stale_count: 10, + archive_count: 10, + }), thresholds) + expect(stale.reasons).toContain('archive_stale') + + const ramp = evaluateQueueHealth(createMetrics({ + archive_recent_count: DEFAULT_ARCHIVE_RECENT_THRESHOLD + 1, + }), thresholds) + expect(ramp.reasons).toContain('archive_ramping') + expect(ramp.reason_details.archive_ramping).toContain('archived in the last') + }) + + it.concurrent('fails when queue or archive tables are missing', () => { + const result = evaluateQueueHealth(createMetrics({ + queue_table_exists: false, + archive_table_exists: false, + }), thresholds) + + expect(result.status).toBe('ko') + expect(result.reasons).toEqual(expect.arrayContaining([ + 'missing_queue_table', + 'missing_archive_table', + ])) + }) + + it.concurrent('fails when queue depth exceeds the threshold', () => { + const result = evaluateQueueHealth(createMetrics({ + queue_count: DEFAULT_QUEUE_DEPTH_THRESHOLD + 1, + }), thresholds) + + expect(result.status).toBe('ko') + expect(result.reasons).toContain('queue_depth_exceeded') + expect(result.reason_details.queue_depth_exceeded).toContain(String(DEFAULT_QUEUE_DEPTH_THRESHOLD)) + }) + + it.concurrent('documents healthy and unhealthy criteria', () => { + const criteria = buildQueueHealthCriteria(thresholds) + expect(criteria.never_read_stale.healthy_when).toContain('read_ct=0') + expect(criteria.archive_stale.unhealthy_when).toContain('ramping') + expect(criteria.stuck_high_read_ct.threshold).toBe(STUCK_READ_CT_THRESHOLD) + }) +})