fix(ci): stop shard 5/6 flakes from PostgREST upstream errors

scripts/repro-shard5-postgrest-flakes.ts

@@ -0,0 +1,172 @@
+/**
+ * Reproduce backend shard 5/6 PostgREST/Kong upstream flakes without Vitest retries.
+ *
+ * CI evidence:
+ * - stats createAppVersions via PostgREST → "An invalid response was received from the upstream server"
+ * - organization-api capgkey auth via PostgREST → intermittent 401 invalid_apikey
+ *
+ * This script forces the upstream failure mode by pausing the local PostgREST
+ * container during the old-path hammer, then proves direct SQL still succeeds.
+ *
+ * Usage (Supabase must be running):
+ *   bun run supabase:with-env -- bun scripts/repro-shard5-postgrest-flakes.ts
+ */
+import { randomUUID } from 'node:crypto'
+import { spawnSync } from 'node:child_process'
+import { createClient } from '@supabase/supabase-js'
+import pg from 'pg'
+
+const SUPABASE_URL = process.env.SUPABASE_URL
+const SERVICE_KEY = process.env.SUPABASE_SERVICE_KEY
+  ?? process.env.SUPABASE_SERVICE_ROLE_KEY
+  ?? process.env.SERVICE_ROLE_KEY
+const DB_URL = process.env.SUPABASE_DB_URL ?? process.env.DB_URL
+
+if (!SUPABASE_URL || !SERVICE_KEY || !DB_URL) {
+  console.error('Missing SUPABASE_URL / service key / DB URL. Use: bun run supabase:with-env -- bun scripts/repro-shard5-postgrest-flakes.ts')
+  process.exit(2)
+}
+
+const ORG_ID = '046a36ac-e03c-4590-9257-bd6c9dba9ee8'
+const USER_ID = '6aa76066-55ef-4238-ade6-0b32334a4097'
+const PARALLEL = Number(process.env.REPRO_PARALLEL ?? 20)
+const REQUEST_TIMEOUT_MS = Number(process.env.REPRO_TIMEOUT_MS ?? 2000)
+
+const supabase = createClient(SUPABASE_URL, SERVICE_KEY, {
+  auth: { persistSession: false },
+  global: {
+    fetch: (input, init) => fetch(input, {
+      ...init,
+      signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
+    }),
+  },
+})
+const pool = new pg.Pool({ connectionString: DB_URL, max: 10 })
+
+function findPostgrestContainer(): string | null {
+  const listed = spawnSync('docker', ['ps', '--format', '{{.Names}}'], { encoding: 'utf8' })
+  if (listed.status !== 0)
+    return null
+  const names = listed.stdout.split('\n').map(s => s.trim()).filter(Boolean)
+  return names.find(name => name.includes('rest') && name.includes('capgo'))
+    ?? names.find(name => name.includes('rest'))
+    ?? null
+}
+
+function dockerAction(action: 'pause' | 'unpause', container: string) {
+  const result = spawnSync('docker', [action, container], { encoding: 'utf8' })
+  if (result.status !== 0) {
+    throw new Error(`docker ${action} ${container} failed: ${result.stderr || result.stdout}`)
+  }
+}
+
+async function ensureApp(appId: string) {
+  await pool.query(
+    `INSERT INTO public.apps (app_id, icon_url, name, owner_org, user_id)
+     VALUES ($1, '', $1, $2::uuid, $3::uuid)
+     ON CONFLICT (app_id) DO NOTHING`,
+    [appId, ORG_ID, USER_ID],
+  )
+}
+
+async function cleanupApp(appId: string) {
+  await pool.query('DELETE FROM public.app_versions WHERE app_id = $1', [appId])
+  await pool.query('DELETE FROM public.apps WHERE app_id = $1', [appId])
+}
+
+async function postgrestCreateVersion(appId: string, version: string) {
+  try {
+    const { data, error } = await supabase.from('app_versions').upsert({
+      app_id: appId,
+      name: version,
+      owner_org: ORG_ID,
+    }, {
+      onConflict: 'app_id,name',
+    }).select('id,name').single()
+    if (error || !data)
+      return { ok: false as const, error: error?.message ?? 'no data' }
+    return { ok: true as const }
+  }
+  catch (error) {
+    const message = error instanceof Error ? error.message : String(error)
+    return { ok: false as const, error: message }
+  }
+}
+
+async function sqlCreateVersion(appId: string, version: string) {
+  const result = await pool.query(
+    `INSERT INTO public.app_versions (app_id, name, owner_org)
+     VALUES ($1, $2, $3::uuid)
+     ON CONFLICT (name, app_id) DO UPDATE SET updated_at = now()
+     RETURNING id, name`,
+    [appId, version, ORG_ID],
+  )
+  if (!result.rows[0])
+    return { ok: false as const, error: 'no data' }
+  return { ok: true as const }
+}
+
+async function runBatch(
+  label: string,
+  fn: (appId: string, version: string) => Promise<{ ok: boolean, error?: string }>,
+) {
+  const appId = `com.repro.shard5.${randomUUID().slice(0, 8)}`
+  await ensureApp(appId)
+  const results = await Promise.all(
+    Array.from({ length: PARALLEL }, (_, i) => {
+      const version = `1.0.0-repro-${i}-${randomUUID().slice(0, 8)}`
+      return fn(appId, version)
+    }),
+  )
+  const failures = results.filter(r => !r.ok)
+  const errors = new Map<string, number>()
+  for (const failure of failures) {
+    const key = failure.error ?? 'unknown'
+    errors.set(key, (errors.get(key) ?? 0) + 1)
+  }
+  console.log(`\n[${label}] failures=${failures.length}/${PARALLEL} (${((failures.length / PARALLEL) * 100).toFixed(1)}%)`)
+  for (const [message, count] of [...errors.entries()].sort((a, b) => b[1] - a[1]).slice(0, 5))
+    console.log(`  - ${count}x ${message}`)
+  await cleanupApp(appId)
+  return failures.length
+}
+
+const container = findPostgrestContainer()
+if (!container) {
+  console.error('Could not find local Postgrest docker container')
+  process.exit(2)
+}
+console.log(`Using PostgREST container: ${container}`)
+
+let postgrestFailures = 0
+try {
+  dockerAction('pause', container)
+  await Bun.sleep(300)
+  postgrestFailures = await runBatch('PostgREST upsert while upstream paused (old path)', postgrestCreateVersion)
+}
+finally {
+  try {
+    dockerAction('unpause', container)
+  }
+  catch {
+    // already running
+  }
+  await Bun.sleep(300)
+}
+
+const sqlFailures = await runBatch('Direct SQL upsert (fixed path)', sqlCreateVersion)
+
+await pool.end()
+
+if (postgrestFailures < PARALLEL) {
+  console.error('\nExpected PostgREST path to fail 100% while upstream is paused.')
+  process.exit(1)
+}
+
+if (sqlFailures > 0) {
+  console.error('\nSQL path failed — fix is incomplete.')
+  process.exit(1)
+}
+
+console.log('\nRepro locked: PostgREST path fails 100% on upstream outage; direct SQL stays green.')
+process.exit(0)

supabase/functions/_backend/public/organization/put.ts

@@ -250,25 +250,38 @@ function buildUpdateFields(body: OrganizationPutBody, sanitizedName?: string) {
 }
 
 async function sanitizeOrgNameForSync(
-  supabase: ReturnType<typeof supabaseApikey>,
+  c: Context<MiddlewareKeyVariables>,
   name: string,
 ) {
-  const { data, error } = await supabase.rpc('strip_html', { input: name })
+  // Direct SQL avoids Kong/PostgREST upstream flakes under parallel test load.
+  const pgPool = getPgClient(c)
+  let client: PgTransactionClient | null = null
+  try {
+    client = await pgPool.connect() as PgTransactionClient
+    const result = await client.query<{ strip_html: string | null }>(
+      'SELECT public.strip_html($1) AS strip_html',
+      [name],
+    )
+    const data = result.rows[0]?.strip_html
+    if (data === null || data === undefined) {
+      throw simpleError('cannot_update_org', 'Cannot update org', {
+        error: 'cannot_sanitize_org_name',
+      })
+    }
 
-  if (error || data === null) {
-    throw simpleError('cannot_update_org', 'Cannot update org', {
-      error: error?.message ?? 'cannot_sanitize_org_name',
-    })
-  }
+    const sanitizedName = data.trim()
+    if (!sanitizedName) {
+      throw simpleError('invalid_body', 'Invalid body', {
+        error: 'sanitized_name_empty',
+      })
+    }
 
-  const sanitizedName = data.trim()
-  if (!sanitizedName) {
-    throw simpleError('invalid_body', 'Invalid body', {
-      error: 'sanitized_name_empty',
-    })
+    return sanitizedName
+  }
+  finally {
+    client?.release()
+    await closeClient(c, pgPool)
   }
-
-  return sanitizedName
 }
 
 async function enforceSelf2faRequirement(authUserId: string, c: Context<MiddlewareKeyVariables>) {
@@ -361,20 +374,28 @@ function buildExpectedCurrentFields(
 }
 
 async function getOrgForNameSync(
-  supabase: ReturnType<typeof supabaseApikey>,
+  c: Context<MiddlewareKeyVariables>,
   orgId: string,
 ): Promise<OrgRow> {
-  const { error, data } = await supabase
-    .from('orgs')
-    .select('*')
-    .eq('id', orgId)
-    .single()
-
-  if (error) {
-    throw simpleError('cannot_get_org', 'Cannot get org', { error: error.message })
+  // Direct SQL avoids Kong/PostgREST upstream flakes under parallel test load.
+  const pgPool = getPgClient(c)
+  let client: PgTransactionClient | null = null
+  try {
+    client = await pgPool.connect() as PgTransactionClient
+    const result = await client.query<OrgRow>(
+      'SELECT * FROM public.orgs WHERE id = $1::uuid LIMIT 1',
+      [orgId],
+    )
+    const data = result.rows[0]
+    if (!data) {
+      throw simpleError('cannot_get_org', 'Cannot get org', { error: 'org_not_found' })
+    }
+    return data
+  }
+  finally {
+    client?.release()
+    await closeClient(c, pgPool)
   }
-
-  return data
 }
 
 function getErrorDetail(error: unknown) {
@@ -417,12 +438,12 @@ export async function put(
   validateMaxExpirationDays(body.max_apikey_expiration_days)
   validateRequiredEncryptionKey(body.required_encryption_key)
   const sanitizedOrgName = body.name !== undefined
-    ? await sanitizeOrgNameForSync(supabase, body.name)
+    ? await sanitizeOrgNameForSync(c, body.name)
     : undefined
   const updateFields = buildUpdateFields(body, sanitizedOrgName)
   const shouldSyncStripeName = body.name !== undefined
   const currentOrg = shouldSyncStripeName
-    ? await getOrgForNameSync(supabase, body.orgId)
+    ? await getOrgForNameSync(c, body.orgId)
     : null
 
   const dataOrg: Database['public']['Tables']['orgs']['Row'] = await updateOrg(c, auth, body.orgId, updateFields, {

supabase/functions/_backend/utils/hono_middleware.ts

@@ -82,9 +82,11 @@ async function checkKeyPg(
       return null
     }
 
-    // Convert to the expected format
+    // Convert to the expected format.
+    // drizzle execute can return numeric ids as strings; keep number so
+    // authApikey.id === existingApikey.id self-update checks work.
     return {
-      id: apiKey.id,
+      id: Number(apiKey.id),
       created_at: apiKey.created_at,
       user_id: apiKey.user_id,
       key: apiKey.key,
@@ -449,7 +451,11 @@ async function foundAPIKey(c: Context, capgkeyString: string) {
   const subkey_id = await getSubkeyId(c)
 
   cloudlog({ requestId: c.get('requestId'), message: 'Capgkey provided', capgkeyPrefix: maskSecret(capgkeyString) })
-  const apikey = await resolveApiKey(c, capgkeyString, false)
+  // Prefer direct Postgres on primary. PostgREST/Kong under parallel Vitest load
+  // returns upstream errors that were previously misclassified as invalid_apikey 401
+  // (flaky organization-api on backend shard 5/6). readOnly=false avoids replica lag
+  // right after key creation.
+  const apikey = await resolveApiKey(c, capgkeyString, true, false)
   if (!apikey) {
     cloudlog({ requestId: c.get('requestId'), message: 'Invalid apikey', capgkeyPrefix: maskSecret(capgkeyString) })
     // Record failed auth attempt - await to ensure accurate counting

supabase/functions/_backend/utils/supabase.ts

@@ -1,13 +1,14 @@
 import type { SupabaseClient } from '@supabase/supabase-js'
 import type { Context } from 'hono'
+import { HTTPException } from 'hono/http-exception'
 import type { BillingPlanBentoState } from './billing_bento_tags.ts'
 import type { AuthInfo } from './hono.ts'
 import type { Database } from './supabase.types.ts'
 import type { DeviceWithoutCreatedAt, NativeVersionUsage, Order, ReadDevicesParams, ReadStatsInsightsParams, ReadStatsParams, StatsInsightsResult, StatsMetadata, VersionUsage, VersionUsageChannel } from './types.ts'
 import { createClient } from '@supabase/supabase-js'
 import { buildBillingPlanBentoTags } from './billing_bento_tags.ts'
 import { buildNormalizedDeviceForWrite, hasComparableDeviceChanged, nullableString } from './deviceComparison.ts'
-import { simpleError } from './hono.ts'
+import { simpleError, quickError } from './hono.ts'
 import { cloudlog, cloudlogErr } from './logging.ts'
 import { closeClient, getPgClient } from './pg.ts'
 import { emptyStatsInsights, normalizeStatsInsightsResult } from './statsInsights.ts'
@@ -1759,7 +1760,17 @@ export async function checkKey(c: Context, authorization: string | undefined, su
       .rpc('find_apikey_by_value', { key_value: authorization })
       .single()
 
-    if (error || !data) {
+    if (error) {
+      // Kong/PostgREST overload must not look like a bad key (flaky 401s in CI).
+      const message = error.message ?? ''
+      if (message.includes('invalid response was received from the upstream server')) {
+        cloudlog({ requestId: c.get('requestId'), message: 'Apikey lookup upstream failure', authorizationPrefix: authorization?.substring(0, 8), error })
+        throw quickError(503, 'upstream_unavailable', 'Upstream unavailable', { error: message })
+      }
+      cloudlog({ requestId: c.get('requestId'), message: 'Invalid apikey', authorizationPrefix: authorization?.substring(0, 8), error })
+      return null
+    }
+    if (!data) {
       cloudlog({ requestId: c.get('requestId'), message: 'Invalid apikey', authorizationPrefix: authorization?.substring(0, 8), error })
       return null
     }
@@ -1773,6 +1784,8 @@ export async function checkKey(c: Context, authorization: string | undefined, su
     return data
   }
   catch (error) {
+    if (error instanceof HTTPException)
+      throw error
     cloudlog({ requestId: c.get('requestId'), message: 'checkKey error', error })
     return null
   }