diff --git a/cloudflare_workers/api/index.ts b/cloudflare_workers/api/index.ts index 73139ba59c..c828f4506a 100644 --- a/cloudflare_workers/api/index.ts +++ b/cloudflare_workers/api/index.ts @@ -68,6 +68,7 @@ import { app as cron_app_fame } from '../../supabase/functions/_backend/triggers import { app as cron_clean_orphan_images } from '../../supabase/functions/_backend/triggers/cron_clean_orphan_images.ts' import { app as cron_clear_versions } from '../../supabase/functions/_backend/triggers/cron_clear_versions.ts' import { app as cron_email } from '../../supabase/functions/_backend/triggers/cron_email.ts' +import { app as cron_onboarding_refresh_apps } from '../../supabase/functions/_backend/triggers/cron_onboarding_refresh_apps.ts' import { app as cron_reconcile_build_status } from '../../supabase/functions/_backend/triggers/cron_reconcile_build_status.ts' import { app as cron_rollout_auto_pause } from '../../supabase/functions/_backend/triggers/cron_rollout_auto_pause.ts' import { app as cron_stat_app } from '../../supabase/functions/_backend/triggers/cron_stat_app.ts' @@ -224,6 +225,7 @@ appTriggers.route('/cron_stat_app', cron_stat_app) appTriggers.route('/cron_stat_org', cron_stat_org) appTriggers.route('/cron_sync_sub', cron_sync_sub) appTriggers.route('/cron_rollout_auto_pause', cron_rollout_auto_pause) +appTriggers.route('/cron_onboarding_refresh_apps', cron_onboarding_refresh_apps) appTriggers.route('/queue_consumer', queue_consumer) appTriggers.route('/send_email', send_email) appTriggers.route('/webhook_delivery', webhook_delivery) diff --git a/read_replicate/schema_replicate.catalog.json b/read_replicate/schema_replicate.catalog.json index 01120db85f..3260623685 100644 --- a/read_replicate/schema_replicate.catalog.json +++ b/read_replicate/schema_replicate.catalog.json @@ -2265,6 +2265,13 @@ "table": "apps", "valid": true }, + { + "constraintOwned": false, + "definition": "CREATE INDEX idx_apps_onboarding_queued_refresh_at ON public.apps USING btree (COALESCE((onboarding ->> 'queued_refresh_at'::text), ''::text), COALESCE((onboarding ->> 'refreshed_at'::text), ''::text), app_id)", + "name": "idx_apps_onboarding_queued_refresh_at", + "table": "apps", + "valid": true + }, { "constraintOwned": false, "definition": "CREATE INDEX idx_apps_onboarding_refreshed_at ON public.apps USING btree (COALESCE((onboarding ->> 'refreshed_at'::text), ''::text), app_id)", diff --git a/read_replicate/schema_replicate.sql b/read_replicate/schema_replicate.sql index b4b7d47616..35d42fe201 100644 --- a/read_replicate/schema_replicate.sql +++ b/read_replicate/schema_replicate.sql @@ -891,6 +891,13 @@ CREATE INDEX idx_apps_onboarding_login_creator ON public.apps USING btree (((onb CREATE INDEX idx_apps_onboarding_ota_stage ON public.apps USING btree (((((onboarding -> 'features'::text) -> 'ota'::text) ->> 'stage'::text))); +-- +-- Name: idx_apps_onboarding_queued_refresh_at; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX idx_apps_onboarding_queued_refresh_at ON public.apps USING btree (COALESCE((onboarding ->> 'queued_refresh_at'::text), ''::text), COALESCE((onboarding ->> 'refreshed_at'::text), ''::text), app_id); + + -- -- Name: idx_apps_onboarding_refreshed_at; Type: INDEX; Schema: public; Owner: - -- diff --git a/src/types/supabase.types.ts b/src/types/supabase.types.ts index 615943f568..b36e4e7355 100644 --- a/src/types/supabase.types.ts +++ b/src/types/supabase.types.ts @@ -5380,10 +5380,6 @@ export type Database = { Args: { p_user_id: string } Returns: string } - refresh_app_onboarding_progress: { - Args: { p_batch_size?: number } - Returns: number - } refresh_app_rollout_channel_count_for_app: { Args: { p_app_id: string } Returns: undefined diff --git a/supabase/functions/_backend/triggers/cron_onboarding_refresh_apps.ts b/supabase/functions/_backend/triggers/cron_onboarding_refresh_apps.ts new file mode 100644 index 0000000000..1139cb96a6 --- /dev/null +++ b/supabase/functions/_backend/triggers/cron_onboarding_refresh_apps.ts @@ -0,0 +1,22 @@ +import type { MiddlewareKeyVariables } from '../utils/hono.ts' +import { Hono } from 'hono/tiny' +import { onboardingRefreshBody, refreshAppOnboardingBatch } from '../utils/app_onboarding_refresh.ts' +import { BRES, middlewareAPISecret, parseBody, quickError } from '../utils/hono.ts' +import { cloudlog } from '../utils/logging.ts' +import { closeClient, getDrizzleClient, getPgClient } from '../utils/pg.ts' + +export const app = new Hono() +app.post('/', middlewareAPISecret, async (c) => { + const parsed = onboardingRefreshBody.safeParse(await parseBody(c)) + if (!parsed.success || new Set(parsed.data.appIds).size !== parsed.data.appIds.length) + throw quickError(400, 'invalid_body', 'Invalid onboarding refresh batch') + const pool = getPgClient(c) + try { + const refreshed = await refreshAppOnboardingBatch(getDrizzleClient(pool), parsed.data) + cloudlog({ requestId: c.get('requestId'), message: 'onboarding refresh batch finished', requested: parsed.data.appIds.length, refreshed }) + return c.json(BRES) + } + finally { + await closeClient(c, pool) + } +}) diff --git a/supabase/functions/_backend/triggers/queue_consumer.ts b/supabase/functions/_backend/triggers/queue_consumer.ts index 411b445c5c..3274e2856e 100644 --- a/supabase/functions/_backend/triggers/queue_consumer.ts +++ b/supabase/functions/_backend/triggers/queue_consumer.ts @@ -1,14 +1,15 @@ import type { Context } from 'hono' import type { MiddlewareKeyVariables } from '../utils/hono.ts' import type { Database } from '../utils/supabase.types.ts' -import { z } from 'zod' import { Hono } from 'hono/tiny' -// --- Worker logic imports --- -import { integerLikeSchema, safeParseSchema } from '../utils/schema_validation.ts' +import { z } from 'zod' +import { ONBOARDING_MESSAGES_PER_MINUTE } from '../utils/app_onboarding_refresh.ts' import { sendDiscordAlert } from '../utils/discord.ts' import { BRES, middlewareAPISecret, parseBody, simpleError } from '../utils/hono.ts' import { cloudlog, cloudlogErr, serializeError } from '../utils/logging.ts' import { closeClient, getPgClient } from '../utils/pg.ts' +// --- Worker logic imports --- +import { integerLikeSchema, safeParseSchema } from '../utils/schema_validation.ts' import { backgroundTask, getEnv, WAIT_FOR_COMPLETION_HEADER } from '../utils/utils.ts' import { updateManifestSize } from './on_manifest_create.ts' @@ -277,6 +278,8 @@ function prepareQueueHttpBody(functionName: string, body: Record, queueName: string, msgIds: number[]) { try { @@ -1096,7 +1102,11 @@ async function mass_edit_queue_messages_cf_ids( // --- Hono app setup --- function shouldRunQueueSyncInBackground(queueName: string): boolean { - return queueName !== 'on_manifest_create' + return queueName !== 'on_manifest_create' && !isOnboardingQueue(queueName) +} + +function isOnboardingQueue(queueName: string): boolean { + return queueName === 'cron_onboarding_refresh_apps' } async function runQueueSync( diff --git a/supabase/functions/_backend/utils/app_onboarding_refresh.ts b/supabase/functions/_backend/utils/app_onboarding_refresh.ts new file mode 100644 index 0000000000..4ca03ca63d --- /dev/null +++ b/supabase/functions/_backend/utils/app_onboarding_refresh.ts @@ -0,0 +1,144 @@ +import type { getDrizzleClient } from './pg.ts' +import { sql } from 'drizzle-orm' +import { z } from 'zod' + +export const ONBOARDING_APPS_PER_MESSAGE = 25 +export const ONBOARDING_MESSAGES_PER_MINUTE = 4 +export const onboardingRefreshBody = z.object({ + appIds: z.array(z.string().min(1)).min(1).max(ONBOARDING_APPS_PER_MESSAGE), + queuedAt: z.iso.datetime(), +}) + +interface OnboardingSignals extends Record { + app_id: string + last_device_at: Date | null + has_app_store: boolean | null + has_testflight: boolean | null + has_play_unknown: boolean | null + has_native: boolean | null + has_install_source: boolean | null + first_bundle_at: Date | null + last_bundle_at: Date | null + first_install_at: Date | null + last_install_at: Date | null + first_build_at: Date | null + first_success_at: Date | null + last_build_at: Date | null +} + +export async function refreshAppOnboardingBatch( + database: Pick, 'transaction'>, + body: z.infer, +) { + return database.transaction(async (tx) => { + await tx.execute(sql`SELECT + pg_catalog.set_config('statement_timeout', '35s', true), + pg_catalog.set_config('lock_timeout', '5s', true), + pg_catalog.set_config('TimeZone', 'UTC', true) + `) + + // Avoid reading large signal tables for messages already covered by a + // later refresh. Recheck this after locking because another worker may win. + const { rows: dueApps } = await tx.execute<{ app_id: string }>(sql` + SELECT app_id FROM public.apps + WHERE app_id = ANY(${sql.param(body.appIds)}::varchar[]) + AND COALESCE(onboarding->>'refreshed_at', '') < ${body.queuedAt} + ORDER BY app_id + `) + if (!dueApps.length) + return 0 + + const dueIds = dueApps.map(app => app.app_id) + // Preserve the old cron's PostgreSQL evidence and date precision. Each + // lateral aggregate starts from an indexed app_id and returns one row. + const { rows: signals } = await tx.execute(sql` + SELECT batch.app_id, + d.last_device_at, d.has_app_store, d.has_testflight, + d.has_play_unknown, d.has_native, d.has_install_source, + v.first_bundle_at, v.last_bundle_at, + dv.first_install_at, dv.last_install_at, + br.first_build_at, br.first_success_at, br.last_build_at + FROM pg_catalog.unnest(${sql.param(dueIds)}::varchar[]) AS batch(app_id) + LEFT JOIN LATERAL ( + SELECT + bool_or(install_source = 'app_store') AS has_app_store, + bool_or(install_source = 'testflight') AS has_testflight, + bool_or(install_source IN ('google_play', 'amazon_appstore', 'samsung_galaxy_store', 'huawei_appgallery')) AS has_play_unknown, + bool_or(is_prod IS TRUE AND is_emulator IS NOT TRUE) AS has_native, + bool_or(install_source IS NOT NULL) AS has_install_source, + max(updated_at) AS last_device_at + FROM public.devices + WHERE app_id = batch.app_id + AND (install_source IS NOT NULL OR (is_prod IS TRUE AND is_emulator IS NOT TRUE)) + ) d ON true + LEFT JOIN LATERAL ( + SELECT min(created_at) AS first_bundle_at, max(created_at) AS last_bundle_at + FROM public.app_versions + WHERE app_id = batch.app_id AND deleted IS NOT TRUE + AND name IS DISTINCT FROM 'builtin' AND name IS DISTINCT FROM 'unknown' + ) v ON true + LEFT JOIN LATERAL ( + SELECT min(date)::timestamptz AS first_install_at, + max(date)::timestamptz AS last_install_at + FROM public.daily_version + WHERE app_id = batch.app_id AND COALESCE(install, 0) > 0 + ) dv ON true + LEFT JOIN LATERAL ( + SELECT min(created_at) AS first_build_at, + min(completed_at) FILTER (WHERE status IN ('succeeded', 'released')) AS first_success_at, + max(COALESCE(completed_at, created_at)) AS last_build_at + FROM public.build_requests WHERE app_id = batch.app_id + ) br ON true + ORDER BY batch.app_id + `) + + await tx.execute(sql` + SELECT app_id FROM public.apps + WHERE app_id = ANY(${sql.param(dueIds)}::varchar[]) + ORDER BY app_id FOR UPDATE + `) + const result = await tx.execute(sql` + WITH signals AS ( + SELECT * FROM pg_catalog.jsonb_to_recordset(${JSON.stringify(signals)}::jsonb) AS s( + app_id varchar, last_device_at timestamptz, + has_app_store boolean, has_testflight boolean, has_play_unknown boolean, + has_native boolean, has_install_source boolean, + first_bundle_at timestamptz, last_bundle_at timestamptz, + first_install_at timestamptz, last_install_at timestamptz, + first_build_at timestamptz, first_success_at timestamptz, last_build_at timestamptz + ) + ) + UPDATE public.apps a SET onboarding = pg_catalog.jsonb_set( + pg_catalog.jsonb_set( + a.onboarding, '{features}', + COALESCE(a.onboarding->'features', '{}'::jsonb) || pg_catalog.jsonb_build_object( + 'cli_install', public.merge_app_onboarding_feature( + a.onboarding->'features'->'cli_install', s.last_device_at, + s.last_device_at, s.last_device_at, NULL), + 'ota', public.merge_app_onboarding_feature( + a.onboarding->'features'->'ota', s.first_bundle_at, s.first_install_at, + GREATEST(s.last_install_at, s.last_bundle_at), + CASE + WHEN s.has_app_store THEN 'store_live' + WHEN s.has_testflight THEN 'testflight' + WHEN s.has_play_unknown THEN 'play_unknown' + WHEN s.has_native THEN 'native_unknown' + WHEN s.has_install_source THEN 'local_only' + ELSE 'no_device' + END), + 'builder', public.merge_app_onboarding_feature( + a.onboarding->'features'->'builder', s.first_build_at, + s.first_success_at, s.last_build_at, NULL) + ), true + ), '{refreshed_at}', + pg_catalog.to_jsonb(pg_catalog.to_char((now() AT TIME ZONE 'UTC'), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')), + true + ) + FROM signals s + WHERE a.app_id = s.app_id + AND COALESCE(a.onboarding->>'refreshed_at', '') < ${body.queuedAt} + RETURNING a.app_id + `) + return result.rowCount ?? 0 + }) +} diff --git a/supabase/functions/_backend/utils/supabase.types.ts b/supabase/functions/_backend/utils/supabase.types.ts index 615943f568..b36e4e7355 100644 --- a/supabase/functions/_backend/utils/supabase.types.ts +++ b/supabase/functions/_backend/utils/supabase.types.ts @@ -5380,10 +5380,6 @@ export type Database = { Args: { p_user_id: string } Returns: string } - refresh_app_onboarding_progress: { - Args: { p_batch_size?: number } - Returns: number - } refresh_app_rollout_channel_count_for_app: { Args: { p_app_id: string } Returns: undefined diff --git a/supabase/functions/triggers/index.ts b/supabase/functions/triggers/index.ts index 99f1f83226..ae1eff6dbb 100644 --- a/supabase/functions/triggers/index.ts +++ b/supabase/functions/triggers/index.ts @@ -5,6 +5,7 @@ import { app as cron_app_fame } from '../_backend/triggers/cron_app_fame.ts' import { app as cron_clean_orphan_images } from '../_backend/triggers/cron_clean_orphan_images.ts' import { app as cron_clear_versions } from '../_backend/triggers/cron_clear_versions.ts' import { app as cron_email } from '../_backend/triggers/cron_email.ts' +import { app as cron_onboarding_refresh_apps } from '../_backend/triggers/cron_onboarding_refresh_apps.ts' import { app as cron_reconcile_build_status } from '../_backend/triggers/cron_reconcile_build_status.ts' import { app as cron_rollout_auto_pause } from '../_backend/triggers/cron_rollout_auto_pause.ts' import { app as cron_stat_app } from '../_backend/triggers/cron_stat_app.ts' @@ -81,6 +82,7 @@ appGlobal.route('/cron_clear_versions', cron_clear_versions) appGlobal.route('/cron_clean_orphan_images', cron_clean_orphan_images) appGlobal.route('/cron_reconcile_build_status', cron_reconcile_build_status) appGlobal.route('/cron_rollout_auto_pause', cron_rollout_auto_pause) +appGlobal.route('/cron_onboarding_refresh_apps', cron_onboarding_refresh_apps) appGlobal.route('/canceled_org_retention_alerts', canceled_org_retention_alerts) appGlobal.route('/credit_usage_alerts', credit_usage_alerts) appGlobal.route('/credit_usage_posthog', credit_usage_posthog) diff --git a/supabase/migrations/20260919133823_backend_onboarding_refresh.sql b/supabase/migrations/20260919133823_backend_onboarding_refresh.sql new file mode 100644 index 0000000000..3e2e94f07b --- /dev/null +++ b/supabase/migrations/20260919133823_backend_onboarding_refresh.sql @@ -0,0 +1,177 @@ +-- Scheduling state lives alongside the existing feature ledger. Never-queued +-- apps sort first; a stuck message can be re-enqueued after 30 minutes. +CREATE INDEX idx_apps_onboarding_queued_refresh_at +ON public.apps ( + (coalesce(onboarding->>'queued_refresh_at', '')), + (coalesce(onboarding->>'refreshed_at', '')), + app_id +); + +SELECT pgmq.create('cron_onboarding_refresh_apps'); + +CREATE OR REPLACE FUNCTION public.enqueue_app_onboarding_refreshes( + p_limit integer DEFAULT 500 +) +RETURNS integer LANGUAGE plpgsql SECURITY DEFINER SET search_path = '' AS $$ +DECLARE + v_batch record; + v_queued_at text := pg_catalog.to_char((now() AT TIME ZONE 'UTC'), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'); + v_total integer := 0; + v_limit integer := GREATEST(1, LEAST(COALESCE(p_limit, 500), 500)); +BEGIN + IF NOT pg_catalog.pg_try_advisory_xact_lock(pg_catalog.hashtext('app_onboarding_refresh_producer')) THEN + RETURN 0; + END IF; + FOR v_batch IN + WITH candidates AS MATERIALIZED ( + SELECT a.app_id, a.onboarding->>'queued_refresh_at' AS queued_at + FROM public.apps a + WHERE COALESCE(a.onboarding->>'refreshed_at', '') < pg_catalog.to_char((now() - interval '10 minutes') AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"') + AND ( + COALESCE(a.onboarding->>'queued_refresh_at', '') <= COALESCE(a.onboarding->>'refreshed_at', '') + OR a.onboarding->>'queued_refresh_at' < pg_catalog.to_char((now() - interval '30 minutes') AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"') + ) + AND COALESCE(( + SELECT si.status = 'succeeded' OR si.trial_at > now() + OR EXISTS ( + SELECT 1 FROM public.usage_credit_grants g + WHERE g.org_id = a.owner_org AND g.expires_at >= now() + AND g.credits_total > g.credits_consumed + ) + FROM public.orgs o + LEFT JOIN public.stripe_info si ON si.customer_id = o.customer_id + WHERE o.id = a.owner_org + ), false) + ORDER BY COALESCE(a.onboarding->>'queued_refresh_at', ''), COALESCE(a.onboarding->>'refreshed_at', ''), a.app_id + LIMIT v_limit FOR UPDATE OF a SKIP LOCKED + ), + queued AS ( + UPDATE public.apps a + SET onboarding = pg_catalog.jsonb_set(a.onboarding, '{queued_refresh_at}', pg_catalog.to_jsonb(v_queued_at), true) + FROM candidates c WHERE a.app_id = c.app_id + RETURNING a.app_id, c.queued_at + ), + numbered AS ( + SELECT app_id, pg_catalog.row_number() OVER ( + ORDER BY COALESCE(queued_at, ''), app_id + ) - 1 AS ordinal FROM queued + ), batches AS ( + SELECT app_id, ordinal / 25 AS batch FROM numbered + ) + SELECT pg_catalog.array_agg(app_id ORDER BY app_id) AS app_ids + FROM batches GROUP BY batch ORDER BY batch + LOOP + PERFORM pgmq.send('cron_onboarding_refresh_apps', pg_catalog.jsonb_build_object( + 'function_name', 'cron_onboarding_refresh_apps', 'function_type', 'cloudflare', + 'payload', pg_catalog.jsonb_build_object('appIds', v_batch.app_ids, 'queuedAt', v_queued_at))); + v_total := v_total + pg_catalog.cardinality(v_batch.app_ids); + END LOOP; + RETURN v_total; +END; +$$; +ALTER FUNCTION public.enqueue_app_onboarding_refreshes( + integer +) OWNER TO postgres; +REVOKE ALL ON FUNCTION public.enqueue_app_onboarding_refreshes( + integer +) FROM public, +anon, +authenticated; +GRANT EXECUTE ON FUNCTION public.enqueue_app_onboarding_refreshes( + integer +) TO service_role; +COMMENT ON FUNCTION public.enqueue_app_onboarding_refreshes(integer) IS +'Internal producer: at most 500 due apps from paying, trial, or credited orgs. +Updates queued_refresh_at and enqueues batches of at most 25 atomically.'; + +-- Reuse the existing scheduler; do not add a pg_cron job. +UPDATE public.cron_tasks SET + task_type = 'function', + target = 'public.enqueue_app_onboarding_refreshes()', + payload = null, + minute_interval = 10, second_interval = null, hour_interval = null, + run_at_hour = null, run_at_minute = null, batch_size = null, enabled = true, + description = 'Enqueue onboarding app refresh batches every 10 minutes', + updated_at = now() +WHERE name = 'refresh_app_onboarding_progress'; + +-- The scheduled batch RPC has been replaced by the SQL producer. Keep the +-- single-app refresh RPC used by verify_getting_started. +DROP FUNCTION public.refresh_app_onboarding_progress(integer); + +COMMENT ON COLUMN public.apps.onboarding IS +'App onboarding state. The backend refresh worker updates feature success, usage, and stage; setup progress is stored separately under setup. Clients may only set feature started_at through mark_onboarding_feature_started.'; +COMMENT ON FUNCTION public.refresh_one_app_onboarding_progress(varchar) IS +'Internal. Refreshes onboarding features for one app from devices, bundles, daily_version installs, and build_requests when Getting Started is verified. Never called from plugin request paths.'; +INSERT INTO public.cron_tasks ( + name, task_type, target, batch_size, minute_interval, description +) +VALUES +( + 'onboarding_refresh_apps_queue', + 'function_queue', + '["cron_onboarding_refresh_apps"]', + 4, + 1, + 'Consume 4 batches of 25 apps per minute (100 apps maximum)' +); + +CREATE OR REPLACE FUNCTION public.process_function_queue( + "queue_name" text, "batch_size" integer DEFAULT 950 +) RETURNS void +LANGUAGE plpgsql +SET search_path TO '' +AS $$ +DECLARE + calls_needed int; + headers jsonb; + queue_size bigint; + request_timeout_ms int; + url text; + onboarding_queue boolean := queue_name = 'cron_onboarding_refresh_apps'; +BEGIN + EXECUTE pg_catalog.format('SELECT count(*) FROM pgmq.%I', 'q_' || queue_name) + INTO queue_size; + + IF queue_size > 0 THEN + IF onboarding_queue THEN + batch_size := LEAST(batch_size, 4); + END IF; + headers := pg_catalog.jsonb_build_object( + 'Content-Type', 'application/json', + 'apisecret', public.get_apikey() + ); + request_timeout_ms := CASE + WHEN queue_name = 'on_manifest_create' OR onboarding_queue THEN 60000 + ELSE 8000 + END; + url := public.get_db_url() || '/functions/v1/triggers/queue_consumer/sync'; + + calls_needed := LEAST( + pg_catalog.ceil(queue_size / batch_size::double precision)::int, + 10 + ); + + IF onboarding_queue THEN + calls_needed := 1; + END IF; + + FOR i IN 1..calls_needed LOOP + PERFORM net.http_post( + url := url, + headers := headers, + body := pg_catalog.jsonb_build_object( + 'queue_name', queue_name, + 'batch_size', batch_size, + 'wait_for_completion', onboarding_queue + ), + timeout_milliseconds := request_timeout_ms + ); + END LOOP; + END IF; +END; +$$; + + +ALTER FUNCTION public.process_function_queue(text, integer) OWNER TO postgres; +REVOKE ALL ON FUNCTION public.process_function_queue(text, integer) FROM public; diff --git a/tests/app-onboarding-progress.test.ts b/tests/app-onboarding-progress.test.ts index b57f9be2f9..063a0a88b9 100644 --- a/tests/app-onboarding-progress.test.ts +++ b/tests/app-onboarding-progress.test.ts @@ -110,21 +110,12 @@ async function insertDevice(appId: string, deviceId: string, installSource: stri throw error } -async function refreshUntil(appId: string) { - for (let attempt = 0; attempt < 40; attempt++) { - await executeSQL('SELECT public.refresh_app_onboarding_progress(500)') - const { data, error } = await serviceRoleSupabase - .from('apps') - .select('onboarding') - .eq('app_id', appId) - .single() - if (error) - throw error - const ledger = parseAppOnboardingLedger(data.onboarding) - if (ledger.refreshed_at) - return ledger - } - throw new Error(`refresh_app_onboarding_progress never reached ${appId}`) +async function refreshOne(appId: string) { + const rows = await executeSQL<{ onboarding: unknown }>( + 'SELECT public.refresh_one_app_onboarding_progress($1) AS onboarding', + [appId], + ) + return parseAppOnboardingLedger(rows[0]?.onboarding) } beforeAll(async () => { @@ -297,13 +288,8 @@ describe('app onboarding progress', () => { }) it('keeps TestFlight-only apps off store_live', async () => { - const defs = await executeSQL<{ def: string }>( - `SELECT pg_get_functiondef('public.refresh_app_onboarding_progress(integer)'::regprocedure) AS def`, - ) - expect(defs[0]?.def).toContain('INNER JOIN batch ON batch.app_id') - - const testflight = await refreshUntil(APP_TESTFLIGHT) - const store = await refreshUntil(APP_STORE) + const testflight = await refreshOne(APP_TESTFLIGHT) + const store = await refreshOne(APP_STORE) expect(testflight.features?.ota?.stage).toBe('testflight') expect(testflight.features?.ota?.stage).not.toBe('store_live') @@ -356,7 +342,7 @@ describe('app onboarding progress', () => { expect(againError).toBeNull() expect(parseAppOnboardingLedger(again).getting_started_dismissed_at).toBe(firstDismissedAt) - const refreshed = await refreshUntil(APP_RPC) + const refreshed = await refreshOne(APP_RPC) expect(refreshed.getting_started_dismissed_at).toBe(firstDismissedAt) expect(refreshed.features?.cli_install?.started_at).toBeTruthy() }) diff --git a/tests/app-onboarding-refresh.unit.test.ts b/tests/app-onboarding-refresh.unit.test.ts new file mode 100644 index 0000000000..17935c5a97 --- /dev/null +++ b/tests/app-onboarding-refresh.unit.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from 'vitest' +import { __queueConsumerTestUtils__, MAX_QUEUE_READS } from '../supabase/functions/_backend/triggers/queue_consumer.ts' +import { onboardingRefreshBody } from '../supabase/functions/_backend/utils/app_onboarding_refresh.ts' + +describe('onboarding refresh queue contract', () => { + it('accepts at most 25 app IDs with the enqueue timestamp', () => { + const queuedAt = '2026-09-19T12:00:00.000Z' + const appIds = Array.from({ length: 25 }, (_, i) => `com.example.${i}`) + expect(onboardingRefreshBody.safeParse({ appIds, queuedAt }).success).toBe(true) + expect(onboardingRefreshBody.safeParse({ appIds: [...appIds, 'com.example.25'], queuedAt }).success).toBe(false) + expect(onboardingRefreshBody.safeParse({ appIds, batchToken: 'unused' }).success).toBe(false) + }) + + it('dispatches at most 100 apps per minute and awaits acknowledgments', () => { + const u = __queueConsumerTestUtils__ + expect(u.getQueueBatchSize('cron_onboarding_refresh_apps', 950)).toBe(4) + expect(u.getQueueHttpConcurrency('cron_onboarding_refresh_apps')).toBe(4) + expect(u.getQueueHttpTimeoutMs('cron_onboarding_refresh_apps')).toBe(45_000) + expect(u.getQueueVisibilityTimeout('cron_onboarding_refresh_apps')).toBe(120) + expect(u.shouldRunQueueSyncInBackground('cron_onboarding_refresh_apps')).toBe(false) + expect(u.getQueueMaxReads('cron_onboarding_refresh_apps')).toBe(MAX_QUEUE_READS) + expect(MAX_QUEUE_READS).toBe(5) + }) +}) diff --git a/tests/cron-onboarding-refresh.test.ts b/tests/cron-onboarding-refresh.test.ts new file mode 100644 index 0000000000..236ee936f2 --- /dev/null +++ b/tests/cron-onboarding-refresh.test.ts @@ -0,0 +1,181 @@ +import { randomUUID } from 'node:crypto' +import { sql } from 'drizzle-orm' +import { afterAll, describe, expect, it } from 'vitest' +import { refreshAppOnboardingBatch } from '../supabase/functions/_backend/utils/app_onboarding_refresh.ts' +import { getDrizzleClient } from '../supabase/functions/_backend/utils/pg.ts' +import { getPostgresClient } from './test-utils.ts' + +afterAll(async () => (await getPostgresClient()).end()) + +async function fixture(client: any, count: number, billing: 'paid' | 'trial' | 'credits' | 'none' = 'none') { + const orgId = randomUUID() + let customerId = billing === 'paid' || billing === 'trial' ? `cus_onboarding_${orgId}` : null + const owner = (await client.query('SELECT id FROM public.users WHERE email = \'test@capgo.app\'')).rows[0].id + if (customerId) { + await client.query(`INSERT INTO public.stripe_info(customer_id,status,product_id,trial_at) + VALUES ($1,$2,'prod_LQIregjtNduh4q',$3::timestamptz)`, [customerId, billing === 'paid' ? 'succeeded' : 'created', billing === 'trial' ? new Date(Date.now() + 86400000).toISOString() : '2020-01-01T00:00:00Z']) + } + await client.query('INSERT INTO public.orgs(id,created_by,name,management_email,customer_id) VALUES ($1,$2,$3,\'onboarding-refresh@example.com\',$4)', [orgId, owner, `Refresh ${orgId}`, customerId]) + customerId = (await client.query('SELECT customer_id FROM public.orgs WHERE id=$1', [orgId])).rows[0].customer_id + if (billing === 'none' || billing === 'credits') { + await client.query('UPDATE public.stripe_info SET status=\'created\', trial_at=now()-interval \'1 day\' WHERE customer_id=$1', [customerId]) + } + if (billing === 'credits') { + await client.query('INSERT INTO public.usage_credit_grants(org_id,credits_total,credits_consumed,expires_at) VALUES ($1,1,0,now()+interval \'1 day\')', [orgId]) + } + const ids = Array.from({ length: count }, (_, i) => `000.r.${orgId}.${String(i).padStart(4, '0')}`) + await client.query(`INSERT INTO public.apps(app_id,owner_org,name,icon_url,need_onboarding) + SELECT app_id,$1,'Refresh fixture','',false FROM pg_catalog.unnest($2::varchar[]) ids(app_id)`, [orgId, ids]) + return { orgId, customerId, ids } +} + +async function cleanup(client: any, fixtures: Awaited>[]) { + for (const item of fixtures) { + await client.query('DELETE FROM public.daily_version WHERE app_id=ANY($1::varchar[])', [item.ids]) + await client.query('DELETE FROM public.devices WHERE app_id=ANY($1::varchar[])', [item.ids]) + await client.query('DELETE FROM public.build_requests WHERE app_id=ANY($1::varchar[])', [item.ids]) + await client.query('DELETE FROM public.app_versions WHERE app_id=ANY($1::varchar[])', [item.ids]) + await client.query('DELETE FROM public.apps WHERE app_id=ANY($1::varchar[])', [item.ids]) + await client.query('DELETE FROM public.usage_credit_grants WHERE org_id=$1', [item.orgId]) + await client.query('DELETE FROM public.orgs WHERE id=$1', [item.orgId]) + if (item.customerId) + await client.query('DELETE FROM public.stripe_info WHERE customer_id=$1', [item.customerId]) + } +} + +describe('backend onboarding refresh', () => { + it('queues only eligible apps, batches 25, and requeues unrefreshed work after 30 minutes', async () => { + const client = await (await getPostgresClient()).connect() + try { + await client.query('BEGIN') + const excluded = await fixture(client, 1) + const paid = await fixture(client, 26, 'paid') + const trial = await fixture(client, 1, 'trial') + const credited = await fixture(client, 1, 'credits') + const eligible = [...paid.ids, ...trial.ids, ...credited.ids] + const countMessagesFor = async (appId: string) => (await client.query(`SELECT count(*)::int AS count + FROM pgmq.q_cron_onboarding_refresh_apps WHERE message->'payload'->'appIds' ? $1`, [appId])).rows[0].count as number + const first = (await client.query('SELECT public.enqueue_app_onboarding_refreshes(500) AS count')).rows[0].count + expect(first).toBeGreaterThanOrEqual(eligible.length) + const ownMessages = (await client.query(`SELECT message->'payload' AS payload FROM pgmq.q_cron_onboarding_refresh_apps + WHERE message->'payload'->'appIds' ?| $1::text[] ORDER BY msg_id`, [eligible])).rows + const queued = ownMessages.flatMap(row => row.payload.appIds).filter((id: string) => eligible.includes(id)) + expect(queued.sort()).toEqual([...eligible].sort()) + expect(ownMessages.every(row => row.payload.appIds.length <= 25)).toBe(true) + expect(ownMessages.every(row => typeof row.payload.queuedAt === 'string')).toBe(true) + expect(await countMessagesFor(excluded.ids[0])).toBe(0) + await client.query('SELECT public.enqueue_app_onboarding_refreshes(500)') + expect(await countMessagesFor(eligible[0])).toBe(1) + await client.query(`UPDATE public.apps SET onboarding = jsonb_set(onboarding,'{queued_refresh_at}', + to_jsonb(to_char((now() - interval '31 minutes') AT TIME ZONE 'UTC','YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'))) + WHERE app_id=ANY($1::varchar[])`, [eligible]) + await client.query('SELECT public.enqueue_app_onboarding_refreshes(500)') + expect(await countMessagesFor(eligible[0])).toBe(2) + await client.query(`UPDATE public.apps SET onboarding = jsonb_set(onboarding,'{refreshed_at}', + to_jsonb(to_char(now() AT TIME ZONE 'UTC','YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'))) + WHERE app_id=ANY($1::varchar[])`, [eligible]) + await client.query('SELECT public.enqueue_app_onboarding_refreshes(500)') + expect(await countMessagesFor(eligible[0])).toBe(2) + } + finally { + await client.query('ROLLBACK') + client.release() + } + }) + + it('keeps the producer cap at 500 after eligibility filtering', async () => { + const client = await (await getPostgresClient()).connect() + try { + await client.query('BEGIN') + await fixture(client, 501, 'paid') + expect((await client.query('SELECT public.enqueue_app_onboarding_refreshes(999999) AS count')).rows[0].count).toBe(500) + } + finally { + await client.query('ROLLBACK') + client.release() + } + }) + + it('derives old PostgreSQL milestones while preserving setup, custom fields, and replay safety', async () => { + const pool = await getPostgresClient() + const client = await pool.connect() + let item: Awaited> | undefined + let released = false + try { + item = await fixture(client, 2) + const [appId, emptyApp] = item.ids + const onboarding = { + setup: { steps: { add_code: { status: 'done' } } }, + getting_started_dismissed_at: '2026-09-01T00:00:00Z', + custom_field: null, + features: { + custom: { succeeded_at: '2026-08-02T00:00:00Z' }, + cli_install: { succeeded_at: '2026-08-03T00:00:00Z' }, + }, + } + await client.query('UPDATE public.apps SET onboarding=$2::jsonb WHERE app_id=$1', [appId, JSON.stringify(onboarding)]) + await client.query(`INSERT INTO public.devices(app_id,device_id,platform,plugin_version,version_name,install_source,is_prod,is_emulator,updated_at) + VALUES ($1,$2,'ios','7.0.0','1.0.0','testflight',true,false,'2026-09-16T12:00:00Z')`, [appId, `device-${randomUUID()}`]) + await client.query(`INSERT INTO public.app_versions(app_id,owner_org,name,created_at) + VALUES ($1,$2,'first-bundle','2026-08-02T00:00:00Z')`, [appId, item.orgId]) + await client.query(`INSERT INTO public.daily_version(date,app_id,version_name,install) + VALUES ('2026-08-05',$1,'1.0.0',1),('2026-09-16',$1,'1.0.0',1)`, [appId]) + const owner = (await client.query('SELECT created_by FROM public.orgs WHERE id=$1', [item.orgId])).rows[0].created_by + await client.query(`INSERT INTO public.build_requests(app_id,owner_org,requested_by,platform,status,upload_session_key,upload_path,upload_url,upload_expires_at,created_at,completed_at) + VALUES ($1,$2,$3,'android','succeeded','test-only','fixture','https://example.com',now(),'2026-08-03T00:00:00Z','2026-08-04T00:00:00Z')`, [appId, item.orgId, owner]) + client.release() + released = true + const queuedAt = new Date(Date.now() - 60000).toISOString() + expect(await refreshAppOnboardingBatch(getDrizzleClient(pool), { appIds: item.ids, queuedAt })).toBe(2) + const row = (await pool.query('SELECT onboarding FROM public.apps WHERE app_id=$1', [appId])).rows[0].onboarding + expect(row.setup).toEqual(onboarding.setup) + expect(row.getting_started_dismissed_at).toBe(onboarding.getting_started_dismissed_at) + expect(row).toHaveProperty('custom_field', null) + expect(row.features.custom).toEqual(onboarding.features.custom) + expect(row.features.cli_install.succeeded_at).toBe('2026-08-03T00:00:00.000Z') + expect(row.features.cli_install.last_used_at).toBe('2026-09-16T12:00:00.000Z') + expect(row.features.ota).toMatchObject({ + started_at: '2026-08-02T00:00:00.000Z', + succeeded_at: '2026-08-05T00:00:00.000Z', + last_used_at: '2026-09-16T00:00:00.000Z', + stage: 'testflight', + }) + expect(row.features.builder).toMatchObject({ + started_at: '2026-08-03T00:00:00.000Z', + succeeded_at: '2026-08-04T00:00:00.000Z', + }) + const empty = (await pool.query('SELECT onboarding FROM public.apps WHERE app_id=$1', [emptyApp])).rows[0].onboarding + expect(empty.features.ota.stage).toBe('no_device') + expect(await refreshAppOnboardingBatch(getDrizzleClient(pool), { appIds: item.ids, queuedAt })).toBe(0) + } + finally { + if (!released) + client.release() + if (item) + await cleanup(pool, [item]) + } + }) + + it('rolls back feature and checkpoint writes when the transaction fails', async () => { + const pool = await getPostgresClient() + const client = await pool.connect() + const item = await fixture(client, 1) + client.release() + try { + const before = (await pool.query('SELECT onboarding FROM public.apps WHERE app_id=$1', [item.ids[0]])).rows[0].onboarding + const database = getDrizzleClient(pool) + const failingDatabase: Pick = { + transaction: (operation, config) => database.transaction(async (tx) => { + const result = await operation(tx) + await tx.execute(sql`SELECT 1/0`) + return result + }, config), + } + await expect(refreshAppOnboardingBatch(failingDatabase, { appIds: item.ids, queuedAt: new Date(Date.now() - 60000).toISOString() })).rejects.toThrow() + expect((await pool.query('SELECT onboarding FROM public.apps WHERE app_id=$1', [item.ids[0]])).rows[0].onboarding).toEqual(before) + } + finally { + await cleanup(pool, [item]) + } + }) +})