diff --git a/cloudflare_workers/api/index.ts b/cloudflare_workers/api/index.ts index 3f7579e2f8..18874b44d5 100644 --- a/cloudflare_workers/api/index.ts +++ b/cloudflare_workers/api/index.ts @@ -76,6 +76,7 @@ import { app as on_organization_create } from '../../supabase/functions/_backend import { app as on_organization_delete } from '../../supabase/functions/_backend/triggers/on_organization_delete.ts' import { app as on_user_create } from '../../supabase/functions/_backend/triggers/on_user_create.ts' import { app as on_user_delete } from '../../supabase/functions/_backend/triggers/on_user_delete.ts' +import { app as on_user_org_access } from '../../supabase/functions/_backend/triggers/on_user_org_access.ts' import { app as on_user_update } from '../../supabase/functions/_backend/triggers/on_user_update.ts' import { app as on_version_create } from '../../supabase/functions/_backend/triggers/on_version_create.ts' import { app as on_version_delete } from '../../supabase/functions/_backend/triggers/on_version_delete.ts' @@ -191,6 +192,7 @@ appTriggers.route('/on_organization_delete', on_organization_delete) appTriggers.route('/on_user_create', on_user_create) appTriggers.route('/on_user_update', on_user_update) appTriggers.route('/on_user_delete', on_user_delete) +appTriggers.route('/on_user_org_access', on_user_org_access) appTriggers.route('/on_version_create', on_version_create) appTriggers.route('/on_version_update', on_version_update) appTriggers.route('/on_version_delete', on_version_delete) diff --git a/docs/BENTO_EMAIL_PREFERENCES_SETUP.md b/docs/BENTO_EMAIL_PREFERENCES_SETUP.md index ff04b73511..26375916b0 100644 --- a/docs/BENTO_EMAIL_PREFERENCES_SETUP.md +++ b/docs/BENTO_EMAIL_PREFERENCES_SETUP.md @@ -72,6 +72,27 @@ Tag does NOT contain: credit_usage_disabled Tag does NOT contain: onboarding_disabled ``` +#### 3a. First-Organization Recovery + +**Entry event**: `user:registered_without_org` + +**Exit or goal event**: `user:joined_org` + +Capgo adds `onboarding:awaiting_first_org` when a direct registration still has no active organization access. When Capgo observes active direct organization access, it removes that tag and permanently adds `onboarding:first_org_recovery_suppressed` in the same subscriber update. The permanent tag keeps recovery disabled even if Bento applies an older asynchronous registration import afterward. A normalized email-address change permanently adds the same suppression tag to both the old and new Bento aliases so asynchronous subscriber updates can never send this recovery email to a stale address. Deleting a user applies the same permanent suppression and unsubscribes the Bento subscriber before organization-dependent cleanup can return. + +Configure the workflow to enroll each subscriber at most once. Queue delivery is at least once, so both lifecycle events can repeat for the same user. Treat `user:joined_org` as a fact used to exit or complete the workflow, not as another entry trigger; the subscriber-level enrollment rule makes repeated `user:registered_without_org` facts harmless. + +After the workflow delay, add a final decision immediately before the send step. All three conditions must still be true: + +```text +Tag contains: onboarding:awaiting_first_org +Tag does NOT contain: onboarding:first_org_recovery_suppressed +Tag does NOT contain: onboarding_disabled +``` + +Do not check the suppression tag only when the workflow starts. It is a monotonic safety opt-out that can be added while the workflow is waiting. +Keep this recovery email non-transactional so Bento's global unsubscribe state remains a delivery gate. + #### 4. Weekly Statistics **Events**: `user:weekly_stats` @@ -159,6 +180,7 @@ Tag does NOT contain: device_error_disabled After configuring Bento, verify the following: - [ ] Each automation has the correct exclusion filter for its disabled tag +- [ ] The first-organization recovery workflow re-checks `onboarding:awaiting_first_org`, `onboarding:first_org_recovery_suppressed`, and `onboarding_disabled` immediately before sending - [ ] Test by disabling a preference for a test user and confirming they don't receive that email type - [ ] Test by re-enabling the preference and confirming they DO receive that email type - [ ] Verify existing users without the `email_preferences` column still receive emails (tags default to not present = enabled) diff --git a/supabase/functions/_backend/triggers/on_user_create.ts b/supabase/functions/_backend/triggers/on_user_create.ts index c2addc34c2..6e4032ce72 100644 --- a/supabase/functions/_backend/triggers/on_user_create.ts +++ b/supabase/functions/_backend/triggers/on_user_create.ts @@ -1,6 +1,7 @@ import type { MiddlewareKeyVariables } from '../utils/hono.ts' import type { Database } from '../utils/supabase.types.ts' import { Hono } from 'hono/tiny' +import { normalizeBentoEmail, prepareNewUserProvisioning, syncBentoFirstOrgOnUserCreate } from '../utils/bento_first_org.ts' import { BRES, middlewareAPISecret, triggerValidator } from '../utils/hono.ts' import { cloudlog } from '../utils/logging.ts' import { createApiKey } from '../utils/supabase.ts' @@ -12,9 +13,17 @@ export const app = new Hono() app.post('/', middlewareAPISecret, triggerValidator('users', 'INSERT'), async (c) => { const record = c.get('webhookBody') as Database['public']['Tables']['users']['Row'] cloudlog({ requestId: c.get('requestId'), message: 'record', record }) + // Configured Bento failures deliberately fail this queue message. Its + // lifecycle tag writes are idempotent, and retrying avoids silently losing + // the only recovery enrollment for a newly registered user. + const shouldProvisionUser = await prepareNewUserProvisioning(c, record) + if (!shouldProvisionUser) + return c.json(BRES) + await createApiKey(c, record.id) cloudlog({ requestId: c.get('requestId'), message: 'createCustomer stripe' }) - await syncUserPreferenceTags(c, record.email, record) + await syncUserPreferenceTags(c, normalizeBentoEmail(record.email), record) + await syncBentoFirstOrgOnUserCreate(c, record) // "User Joined" should represent a self-signup (technical user expected to onboard), // not an account created by accepting an org invite. await sendEventToTracking(c, { diff --git a/supabase/functions/_backend/triggers/on_user_delete.ts b/supabase/functions/_backend/triggers/on_user_delete.ts index b63245e152..8a8f045706 100644 --- a/supabase/functions/_backend/triggers/on_user_delete.ts +++ b/supabase/functions/_backend/triggers/on_user_delete.ts @@ -3,13 +3,16 @@ import type { MiddlewareKeyVariables } from '../utils/hono.ts' import type { Database } from '../utils/supabase.types.ts' import { Hono } from 'hono/tiny' import { unsubscribeBento } from '../utils/bento.ts' -import { BRES, middlewareAPISecret, triggerValidator } from '../utils/hono.ts' +import { normalizeBentoEmail, suppressAndUnsubscribeDeletedUserRecovery } from '../utils/bento_first_org.ts' +import { BRES, middlewareAPISecret, quickError, triggerValidator } from '../utils/hono.ts' import { cloudlog } from '../utils/logging.ts' import { cancelSubscription } from '../utils/stripe.ts' import { supabaseAdmin } from '../utils/supabase.ts' export const app = new Hono() +const BENTO_USER_DELETE_TIMEOUT_MS = 5_000 + interface RbacBinding { org_id?: string | null principal_id?: string | null @@ -253,7 +256,6 @@ function getSingleSuperAdminOrgs( function buildCleanupPromises( c: Context, orgs: Array<{ customer_id: string | null, management_email: string | null }> | null, - record: Database['public']['Tables']['users']['Row'], ) { const promises: Promise[] = [] if (orgs && orgs.length > 0) { @@ -269,11 +271,51 @@ function buildCleanupPromises( } } - if (record.email) { - promises.push(unsubscribeBento(c, record.email)) + return promises +} + +async function suppressDeletedUserInBento( + c: Context, + record: Database['public']['Tables']['users']['Row'], + signal?: AbortSignal, +) { + if (!record.email) + return true + + const email = normalizeBentoEmail(record.email) + try { + await suppressAndUnsubscribeDeletedUserRecovery(c, email, signal) + return true + } + catch (error) { + logFailure(c, 'Bento user deletion cleanup failed', error) + return false + } +} + +async function suppressDeletedUserInBentoWithinBudget( + c: Context, + record: Database['public']['Tables']['users']['Row'], +) { + const controller = new AbortController() + const timeoutId = setTimeout(() => controller.abort(), BENTO_USER_DELETE_TIMEOUT_MS) + try { + const result = await suppressDeletedUserInBento(c, record, controller.signal) + if (controller.signal.aborted) { + logFailure(c, 'Bento user deletion cleanup timed out') + return false + } + return result } + finally { + clearTimeout(timeoutId) + } +} - return promises +function finishDeleteUserResponse(c: Context, bentoCleanupSucceeded: boolean, resourceCleanupSucceeded: boolean) { + if (!bentoCleanupSucceeded || !resourceCleanupSucceeded) + quickError(500, 'user_delete_cleanup_failed', 'User deletion cleanup failed') + return c.json(BRES) } async function deleteUserImages( @@ -282,28 +324,36 @@ async function deleteUserImages( userId: string, ) { try { - const { data: files } = await supabase + const { data: files, error: listError } = await supabase .storage .from('images') .list(userId) + if (listError) { + logFailure(c, 'user image listing failed', listError) + return false + } if (files && files.length > 0) { const filePaths = files.map(file => `${userId}/${file.name}`) - await supabase + const { error: removeError } = await supabase .storage .from('images') .remove(filePaths) + if (removeError) { + logFailure(c, 'user image deletion failed', removeError) + return false + } cloudlog({ requestId: c.get('requestId'), message: 'deleted user images', count: files.length, user_id: userId }) } + return true } catch (error) { cloudlog({ requestId: c.get('requestId'), message: 'error deleting user images', error, user_id: userId }) + return false } } -// on_user_delete - this is called 30 days before the user is actually deleted -// This function is used to cancel the subscriptions of the user's organizations -async function deleteUser(c: Context, record: Database['public']['Tables']['users']['Row']) { +async function cleanupDeletedUserResources(c: Context, record: Database['public']['Tables']['users']['Row']) { // Process user deletion with timeout protection const startTime = Date.now() const supabase = supabaseAdmin(c) @@ -311,35 +361,29 @@ async function deleteUser(c: Context, record: Database['public']['Tables']['user // 1. Find organizations where this user is the only RBAC super admin const directRbacBindings = await fetchDirectRbacBindings(c, supabase, record.id) - if (!directRbacBindings) { - return c.json(BRES) - } + if (!directRbacBindings) + return false const groupIds = await fetchUserGroupIds(c, supabase, record.id) - if (!groupIds) { - return c.json(BRES) - } + if (!groupIds) + return false const groupRbacBindings = await fetchGroupRbacBindings(c, supabase, groupIds) - if (!groupRbacBindings) { - return c.json(BRES) - } + if (!groupRbacBindings) + return false const orgIds = collectCandidateOrgIds(directRbacBindings, groupRbacBindings, now) - if (orgIds.length === 0) { - return c.json(BRES) - } + if (orgIds.length === 0) + return true // For each org where user is super admin, check if they are the only one const rbacUserAdmins = await fetchRbacUserAdmins(c, supabase, orgIds) - if (!rbacUserAdmins) { - return c.json(BRES) - } + if (!rbacUserAdmins) + return false const rbacGroupAdmins = await fetchRbacGroupAdmins(c, supabase, orgIds) - if (!rbacGroupAdmins) { - return c.json(BRES) - } + if (!rbacGroupAdmins) + return false const activeGroupBindings = rbacGroupAdmins.filter(binding => isBindingActive(binding, now)) const rbacGroupIds = activeGroupBindings @@ -347,9 +391,8 @@ async function deleteUser(c: Context, record: Database['public']['Tables']['user .filter((groupId): groupId is string => Boolean(groupId)) const groupMembers = await fetchGroupMembers(c, supabase, rbacGroupIds) - if (!groupMembers) { - return c.json(BRES) - } + if (!groupMembers) + return false const groupMembersByGroup = buildGroupMembersByGroup(groupMembers) const orgAdminUsers = buildOrgAdminUsers( @@ -361,19 +404,33 @@ async function deleteUser(c: Context, record: Database['public']['Tables']['user const singleSuperAdminOrgs = getSingleSuperAdminOrgs(orgIds, orgAdminUsers, record.id) - if (singleSuperAdminOrgs.length === 0) { - return c.json(BRES) - } + if (singleSuperAdminOrgs.length === 0) + return true - const { data: orgs } = await supabaseAdmin(c) + const { data: orgs, error: orgLookupError } = await supabaseAdmin(c) .from('orgs') .select('id, customer_id, management_email') .in('id', singleSuperAdminOrgs) + if (orgLookupError) { + logFailure(c, 'organization cleanup lookup failed', orgLookupError) + return false + } - const promises = buildCleanupPromises(c, orgs ?? null, record) + const promises = buildCleanupPromises(c, orgs ?? null) promises.push(deleteUserImages(c, supabase, record.id)) - await Promise.all(promises) + let cleanupResults: unknown[] + try { + cleanupResults = await Promise.all(promises) + } + catch (error) { + logFailure(c, 'user resource cleanup failed', error) + return false + } + if (cleanupResults.includes(false)) { + logFailure(c, 'user resource cleanup reported a failure') + return false + } // 4. Track performance metrics const endTime = Date.now() @@ -385,8 +442,19 @@ async function deleteUser(c: Context, record: Database['public']['Tables']['user duration_ms: duration, user_id: record.id, }) + return true +} - return c.json(BRES) +// on_user_delete - this is called 30 days before the user is actually deleted. +// Start provider cleanup and core subscription/image cleanup together so a +// slow Bento request cannot consume the queue handler's entire 15-second +// budget before Stripe cancellation begins. +async function deleteUser(c: Context, record: Database['public']['Tables']['users']['Row']) { + const [bentoCleanupSucceeded, resourceCleanupSucceeded] = await Promise.all([ + suppressDeletedUserInBentoWithinBudget(c, record), + cleanupDeletedUserResources(c, record), + ]) + return finishDeleteUserResponse(c, bentoCleanupSucceeded, resourceCleanupSucceeded) } app.post('/', middlewareAPISecret, triggerValidator('users', 'DELETE'), async (c) => { diff --git a/supabase/functions/_backend/triggers/on_user_org_access.ts b/supabase/functions/_backend/triggers/on_user_org_access.ts new file mode 100644 index 0000000000..d0183a5ef8 --- /dev/null +++ b/supabase/functions/_backend/triggers/on_user_org_access.ts @@ -0,0 +1,17 @@ +import type { Database } from '../utils/supabase.types.ts' +import { z } from 'zod' +import { syncBentoFirstOrgOnRoleBindingWrite } from '../utils/bento_first_org.ts' +import { BRES, createHono, middlewareAPISecret, simpleError, triggerValidator } from '../utils/hono.ts' +import { version } from '../utils/version.ts' + +export const app = createHono('', version) + +app.post('/', middlewareAPISecret, triggerValidator('role_bindings', ['INSERT', 'UPDATE']), async (c) => { + const record = c.get('webhookBody') as Partial + const bindingId = z.uuid().safeParse(record.id) + if (!bindingId.success) + throw simpleError('invalid_payload', 'Invalid role binding id', { id: record.id }) + + await syncBentoFirstOrgOnRoleBindingWrite(c, bindingId.data) + return c.json(BRES) +}) diff --git a/supabase/functions/_backend/triggers/on_user_update.ts b/supabase/functions/_backend/triggers/on_user_update.ts index 897c9947a1..f1ae3013dd 100644 --- a/supabase/functions/_backend/triggers/on_user_update.ts +++ b/supabase/functions/_backend/triggers/on_user_update.ts @@ -1,7 +1,8 @@ import type { MiddlewareKeyVariables } from '../utils/hono.ts' import type { Database } from '../utils/supabase.types.ts' import { Hono } from 'hono/tiny' -import { BRES, middlewareAPISecret, simpleError, triggerValidator } from '../utils/hono.ts' +import { normalizeBentoEmail, syncBentoFirstOrgOnEmailChange } from '../utils/bento_first_org.ts' +import { BRES, middlewareAPISecret, quickError, simpleError, triggerValidator } from '../utils/hono.ts' import { cleanStoredImageMetadata } from '../utils/image.ts' import { cloudlog } from '../utils/logging.ts' import { createApiKey } from '../utils/supabase.ts' @@ -21,8 +22,18 @@ app.post('/', middlewareAPISecret, triggerValidator('users', 'UPDATE'), async (c cloudlog({ requestId: c.get('requestId'), message: 'No id' }) throw simpleError('no_id', 'No id', { record }) } + + const newEmail = normalizeBentoEmail(record.email) + const oldEmail = oldRecord?.email ? normalizeBentoEmail(oldRecord.email) : undefined + if (oldEmail && oldEmail !== newEmail) { + const suppressionResult = await syncBentoFirstOrgOnEmailChange(c, oldEmail, newEmail) + if (suppressionResult === false) + quickError(500, 'bento_first_org_suppression_failed', 'Bento first-organization recovery suppression failed') + } + + await syncUserPreferenceTags(c, newEmail, record, oldRecord, oldEmail) + await createApiKey(c, record.id) - await syncUserPreferenceTags(c, record.email, record, oldRecord, oldRecord?.email) const newImagePath = record.image_url const oldImagePath = oldRecord?.image_url diff --git a/supabase/functions/_backend/utils/bento.ts b/supabase/functions/_backend/utils/bento.ts index 8e796b28ba..42db716610 100644 --- a/supabase/functions/_backend/utils/bento.ts +++ b/supabase/functions/_backend/utils/bento.ts @@ -37,7 +37,7 @@ function getBentoHeaders(c: Context) { } } -async function bentoFetch(c: Context, path: string, siteUuid: string, body: any) { +async function bentoFetch(c: Context, path: string, siteUuid: string, body: any, signal?: AbortSignal) { const headers = getBentoHeaders(c) if (!headers) return null @@ -49,6 +49,7 @@ async function bentoFetch(c: Context, path: string, siteUuid: string, body: any) method: 'POST', headers, body: JSON.stringify(body), + signal, }) if (!response.ok) { @@ -59,6 +60,21 @@ async function bentoFetch(c: Context, path: string, siteUuid: string, body: any) return response.json() } +function acceptedBentoBatchResult(result: unknown, expectedResults: number) { + if (!result || typeof result !== 'object') + return false + + const response = result as { failed?: unknown, results?: unknown } + return response.results === expectedResults && response.failed === 0 +} + +function acceptedBentoCommandResult(result: unknown, expectedResults: number) { + if (!result || typeof result !== 'object') + return false + + return (result as { results?: unknown }).results === expectedResults +} + // Only use this function when a specific member of the organization needs to be tracked in Bento. For organization-level events, use sendNotifToOrgMembers in org_email_notifications.ts which will call trackBentoEvent for each member with an email in the background. export async function trackBentoEvent(c: Context, email: string, data: Record, event: string) { if (!isBentoConfigured(c)) @@ -75,8 +91,8 @@ export async function trackBentoEvent(c: Context, email: string, data: Record 0) { + const res = await bentoFetch(c, 'batch/events', siteUuid, payload) + if (!acceptedBentoBatchResult(res, payload.events.length)) { cloudlogErr({ requestId: c.get('requestId'), message: 'trackBentoEvent', error: res }) return false } @@ -124,6 +140,7 @@ export async function addTagBento(c: Context, email: string, segments: { segment export async function syncBentoSubscriberTags( c: Context, update: { email: string, segments: string[], deleteSegments: string[] } | Array<{ email: string, segments: string[], deleteSegments: string[] }>, + signal?: AbortSignal, ) { if (!isBentoConfigured(c)) return @@ -150,8 +167,8 @@ export async function syncBentoSubscriberTags( for (let i = 0; i < subscribers.length; i += chunkSize) { const chunk = subscribers.slice(i, i + chunkSize) const payload = { subscribers: chunk } - const res = await bentoFetch(c, 'batch/subscribers', siteUuid, payload) as { results?: number, failed?: number, errors?: unknown } - if (res?.failed && res.failed > 0) { + const res = await bentoFetch(c, 'batch/subscribers', siteUuid, payload, signal) + if (!acceptedBentoBatchResult(res, chunk.length)) { cloudlogErr({ requestId: c.get('requestId'), message: 'syncBentoSubscriberTags', error: res }) return false } @@ -164,7 +181,7 @@ export async function syncBentoSubscriberTags( } } -export async function unsubscribeBento(c: Context, email: string) { +export async function unsubscribeBento(c: Context, email: string, signal?: AbortSignal) { if (!isBentoConfigured(c)) return @@ -175,8 +192,12 @@ export async function unsubscribeBento(c: Context, email: string) { email, } - const result = await bentoFetch(c, 'fetch/commands', siteUuid, { command }) + const result = await bentoFetch(c, 'fetch/commands', siteUuid, { command }, signal) + if (!acceptedBentoCommandResult(result, 1)) { + cloudlogErr({ requestId: c.get('requestId'), message: 'unsubscribeBento rejected', error: result }) + return false + } cloudlog({ requestId: c.get('requestId'), message: 'unsubscribeBento', email, result }) return true } diff --git a/supabase/functions/_backend/utils/bento_first_org.ts b/supabase/functions/_backend/utils/bento_first_org.ts new file mode 100644 index 0000000000..1ad34ae9c2 --- /dev/null +++ b/supabase/functions/_backend/utils/bento_first_org.ts @@ -0,0 +1,414 @@ +import type { Context } from 'hono' +import type { MiddlewareKeyVariables } from './hono.ts' +import type { Database } from './supabase.types.ts' +import { syncBentoSubscriberTags, trackBentoEvent, unsubscribeBento } from './bento.ts' +import { quickError } from './hono.ts' +import { closeClient, getPgClient } from './pg.ts' + +export const BENTO_AWAITING_FIRST_ORG_TAG = 'onboarding:awaiting_first_org' +// Permanent safety opt-out: never remove this tag. The Bento recovery workflow +// must require it to be absent immediately before sending a recovery email. +export const BENTO_FIRST_ORG_RECOVERY_SUPPRESSED_TAG = 'onboarding:first_org_recovery_suppressed' +export const BENTO_REGISTERED_WITHOUT_ORG_EVENT = 'user:registered_without_org' +export const BENTO_JOINED_ORG_EVENT = 'user:joined_org' + +export const BENTO_DELETED_USER_OPERATION_TIMEOUT_MS = 2_000 + +interface CurrentRoleBinding { + email: string | null + granted_at: Date + id: string + is_active: boolean + is_direct: boolean + org_id: string | null + principal_id: string + principal_type: string + scope_type: string +} + +export interface FirstOrgRegistrationState { + has_active_direct_org_access: boolean + user_is_recovery_eligible: boolean +} + +export function normalizeBentoEmail(email: string) { + return email.trim().toLowerCase() +} + +function ensureBentoDelivery(result: boolean | undefined, operation: string) { + if (result === false) + quickError(500, 'bento_lifecycle_delivery_failed', 'Bento lifecycle delivery failed', { operation }) +} + +async function runDeletedUserBentoOperation( + operation: string, + parentSignal: AbortSignal | undefined, + startOperation: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + let parentAborted = parentSignal?.aborted ?? false + let timedOut = false + const abortFromParent = () => { + parentAborted = true + controller.abort() + } + if (parentAborted) + controller.abort() + else + parentSignal?.addEventListener('abort', abortFromParent, { once: true }) + + const timeoutId = setTimeout(() => { + timedOut = true + controller.abort() + }, BENTO_DELETED_USER_OPERATION_TIMEOUT_MS) + try { + // Await the aborted fetch's settlement before starting the next mutation; + // a timeout must not leave a detached suppression request behind. + const result = await startOperation(controller.signal) + if (timedOut) + throw new Error(`${operation} timed out`) + if (parentAborted) + throw new Error(`${operation} aborted`) + return result + } + finally { + clearTimeout(timeoutId) + parentSignal?.removeEventListener('abort', abortFromParent) + } +} + +async function setAwaitingFirstOrgTag(c: Context, email: string, awaiting: boolean) { + // Bento imports subscriber updates asynchronously. A terminal transition + // must therefore add the monotonic suppression tag in the same update that + // removes awaiting state: a delayed registration import may re-add awaiting, + // but it can never remove suppression and re-enable the recovery send gate. + const result = await syncBentoSubscriberTags(c, { + email, + segments: awaiting ? [BENTO_AWAITING_FIRST_ORG_TAG] : [BENTO_FIRST_ORG_RECOVERY_SUPPRESSED_TAG], + deleteSegments: awaiting ? [] : [BENTO_AWAITING_FIRST_ORG_TAG], + }) + ensureBentoDelivery(result, awaiting ? 'add_awaiting_first_org_tag' : 'suppress_first_org_recovery') +} + +export async function syncBentoFirstOrgOnEmailChange( + c: Context, + oldEmail: string, + newEmail: string, + signal?: AbortSignal, +) { + const aliases = [...new Set([oldEmail, newEmail].map(normalizeBentoEmail).filter(Boolean))] + const updates = aliases.map(email => ({ + email, + segments: [BENTO_FIRST_ORG_RECOVERY_SUPPRESSED_TAG], + deleteSegments: [BENTO_AWAITING_FIRST_ORG_TAG], + })) + return signal + ? await syncBentoSubscriberTags(c, updates, signal) + : await syncBentoSubscriberTags(c, updates) +} + +export async function suppressAndUnsubscribeDeletedUserRecovery( + c: Context, + email: string, + signal?: AbortSignal, +) { + let suppressionResult: boolean | undefined + let suppressionError: unknown + let suppressionThrew = false + try { + suppressionResult = await runDeletedUserBentoOperation( + 'suppress deleted user recovery', + signal, + signal => syncBentoFirstOrgOnEmailChange(c, email, email, signal), + ) + } + catch (error) { + suppressionError = error + suppressionThrew = true + } + + // Submit unsubscribe after suppression for an identity that is already + // scheduled for deletion, even if suppression delivery throws. + let unsubscribeResult: boolean | undefined + let unsubscribeError: unknown + let unsubscribeThrew = false + try { + unsubscribeResult = await runDeletedUserBentoOperation( + 'unsubscribe deleted user recovery', + signal, + signal => unsubscribeBento(c, email, signal), + ) + } + catch (error) { + unsubscribeError = error + unsubscribeThrew = true + } + + if (suppressionThrew) + throw suppressionError + if (unsubscribeThrew) + throw unsubscribeError + ensureBentoDelivery(suppressionResult, 'suppress_deleted_user_recovery') + ensureBentoDelivery(unsubscribeResult, 'unsubscribe_deleted_user_recovery') +} + +async function reconcileFirstOrgStateAfterBentoMutation( + c: Context, + pgPool: ReturnType, + userId: string, + email: string, +) { + const state = await getFirstOrgDatabaseState(pgPool, userId) + if (!state.user_is_recovery_eligible) + await suppressAndUnsubscribeDeletedUserRecovery(c, email) + return state +} + +async function runBentoMutationWithFirstOrgReconciliation( + c: Context, + pgPool: ReturnType, + userId: string, + email: string, + mutate: () => Promise, +) { + let mutationSucceeded = false + let mutationError: unknown + try { + await mutate() + mutationSucceeded = true + } + catch (error) { + mutationError = error + } + + // Reconcile even after an ambiguous provider failure: Bento may have + // accepted the mutation before Capgo lost the response. + let state: Awaited> + try { + state = await reconcileFirstOrgStateAfterBentoMutation(c, pgPool, userId, email) + } + catch (reconciliationError) { + if (!mutationSucceeded) { + throw new AggregateError( + [mutationError, reconciliationError], + 'Bento mutation and first-organization reconciliation failed', + ) + } + throw reconciliationError + } + if (!mutationSucceeded) + throw mutationError + return state +} + +async function getFirstOrgDatabaseState(pgPool: ReturnType, userId: string) { + const pgClient = await pgPool.connect() + try { + const result = await pgClient.query( + `SELECT + EXISTS ( + SELECT 1 + FROM public.users AS users + WHERE users.id = $1::uuid + AND NOT EXISTS ( + SELECT 1 + FROM public.to_delete_accounts AS deleted + WHERE deleted.account_id = users.id + ) + ) AS user_is_recovery_eligible, + EXISTS ( + SELECT 1 + FROM public.role_bindings + WHERE principal_type = 'user' + AND principal_id = $1::uuid + AND scope_type = 'org' + AND org_id IS NOT NULL + AND is_direct IS TRUE + AND (expires_at IS NULL OR expires_at > pg_catalog.now()) + ) AS has_active_direct_org_access`, + [userId], + ) + return result.rows[0] ?? { + has_active_direct_org_access: false, + user_is_recovery_eligible: false, + } + } + finally { + // General-backend Pools are request-scoped and closeClient intentionally + // does not end them in workerd. Destroy the checked-out socket at the query + // boundary so it cannot survive across Bento I/O or request teardown. + pgClient.release(true) + } +} + +export async function prepareNewUserProvisioning( + c: Context, + user: Database['public']['Tables']['users']['Row'], +) { + const email = normalizeBentoEmail(user.email) + const pgPool = getPgClient(c) + try { + const state = await getFirstOrgDatabaseState(pgPool, user.id) + if (!state.user_is_recovery_eligible) { + await suppressAndUnsubscribeDeletedUserRecovery(c, email) + return false + } + return true + } + finally { + await closeClient(c, pgPool) + } +} + +export async function syncBentoFirstOrgOnUserCreate( + c: Context, + user: Database['public']['Tables']['users']['Row'], +) { + const email = normalizeBentoEmail(user.email) + const pgPool = getPgClient(c) + try { + // Provisioning can overlap account deletion. Read before lifecycle work + // and reconcile after each terminal mutation so recovery stays fail closed. + const postProvisionState = await getFirstOrgDatabaseState(pgPool, user.id) + if (!postProvisionState.user_is_recovery_eligible) { + await suppressAndUnsubscribeDeletedUserRecovery(c, email) + return + } + if (user.created_via_invite || postProvisionState.has_active_direct_org_access) { + await runBentoMutationWithFirstOrgReconciliation( + c, + pgPool, + user.id, + email, + async () => await setAwaitingFirstOrgTag(c, email, false), + ) + return + } + + const finalState = await runBentoMutationWithFirstOrgReconciliation( + c, + pgPool, + user.id, + email, + async () => await setAwaitingFirstOrgTag(c, email, true), + ) + if (!finalState.user_is_recovery_eligible) + return + if (finalState.has_active_direct_org_access) { + await runBentoMutationWithFirstOrgReconciliation( + c, + pgPool, + user.id, + email, + async () => await setAwaitingFirstOrgTag(c, email, false), + ) + return + } + + // This read deliberately follows the final registration-side Bento + // mutation. If deletion committed first, suppress and unsubscribe here; + // if it commits later, the queued deletion worker submits cleanup afterward. + const settledState = await runBentoMutationWithFirstOrgReconciliation( + c, + pgPool, + user.id, + email, + async () => { + const result = await trackBentoEvent(c, email, { + user_id: user.id, + registered_at: user.created_at, + created_via_invite: false, + }, BENTO_REGISTERED_WITHOUT_ORG_EVENT) + ensureBentoDelivery(result, 'registered_without_org_event') + }, + ) + if (settledState.user_is_recovery_eligible && settledState.has_active_direct_org_access) { + await runBentoMutationWithFirstOrgReconciliation( + c, + pgPool, + user.id, + email, + async () => await setAwaitingFirstOrgTag(c, email, false), + ) + } + } + finally { + await closeClient(c, pgPool) + } +} + +export async function syncBentoFirstOrgOnRoleBindingWrite( + c: Context, + roleBindingId: string, +) { + const pgPool = getPgClient(c) + try { + let binding: CurrentRoleBinding | undefined + const pgClient = await pgPool.connect() + try { + const result = await pgClient.query( + `SELECT + binding.id, + binding.principal_type, + binding.principal_id, + binding.scope_type, + binding.org_id, + binding.granted_at, + binding.is_direct, + (binding.expires_at IS NULL OR binding.expires_at > pg_catalog.now()) AS is_active, + users.email + FROM public.role_bindings AS binding + LEFT JOIN public.users AS users + ON users.id = binding.principal_id + WHERE binding.id = $1::uuid + LIMIT 1`, + [roleBindingId], + ) + binding = result.rows[0] + } + finally { + // See hasActiveDirectOrgAccess: destroy this request-scoped socket before + // the handler crosses the network boundary into Bento. + pgClient.release(true) + } + + if ( + !binding + || binding.principal_type !== 'user' + || binding.scope_type !== 'org' + || !binding.org_id + || binding.is_direct !== true + || binding.is_active !== true + || !binding.email + ) { + return + } + + const email = normalizeBentoEmail(binding.email) + const preMutationState = await getFirstOrgDatabaseState(pgPool, binding.principal_id) + if (!preMutationState.user_is_recovery_eligible) { + await suppressAndUnsubscribeDeletedUserRecovery(c, email) + return + } + + await runBentoMutationWithFirstOrgReconciliation( + c, + pgPool, + binding.principal_id, + email, + async () => { + await setAwaitingFirstOrgTag(c, email, false) + const result = await trackBentoEvent(c, email, { + user_id: binding.principal_id, + org_id: binding.org_id, + role_binding_id: binding.id, + joined_at: binding.granted_at.toISOString(), + }, BENTO_JOINED_ORG_EVENT) + ensureBentoDelivery(result, 'joined_org_event') + }, + ) + } + finally { + await closeClient(c, pgPool) + } +} diff --git a/supabase/functions/_backend/utils/hono.ts b/supabase/functions/_backend/utils/hono.ts index d9c890b996..fc43d400b6 100644 --- a/supabase/functions/_backend/utils/hono.ts +++ b/supabase/functions/_backend/utils/hono.ts @@ -173,20 +173,27 @@ export const useCors = cors({ export const honoFactory = createFactory() +type TriggerOperation = 'DELETE' | 'INSERT' | 'UPDATE' + export function triggerValidator( table: keyof Database['public']['Tables'], - type: 'DELETE' | 'INSERT' | 'UPDATE', + type: TriggerOperation | readonly TriggerOperation[], ) { return honoFactory.createMiddleware(async (c, next) => { - const body = await c.req.json | InsertPayload | UpdatePayload>() + const rawBody = await parseBody(c) + if (rawBody === null || typeof rawBody !== 'object' || Array.isArray(rawBody)) + throw simpleError('invalid_payload', 'Expected a JSON object', { body: rawBody }) + + const body = rawBody as DeletePayload | InsertPayload | UpdatePayload + const allowedTypes: readonly TriggerOperation[] = typeof type === 'string' ? [type] : type if (body.table !== String(table)) { cloudlog({ requestId: c.get('requestId'), message: `Not ${String(table)}` }) throw simpleError('table_not_match', 'Not table', { body }) } - if (body.type !== type) { - cloudlog({ requestId: c.get('requestId'), message: `Not ${type}` }) + if (!allowedTypes.includes(body.type)) { + cloudlog({ requestId: c.get('requestId'), message: `Not ${allowedTypes.join(' or ')}` }) throw simpleError('type_not_match', 'Not type', { body }) } diff --git a/supabase/functions/_backend/utils/stripe.ts b/supabase/functions/_backend/utils/stripe.ts index 8f2c2820bb..4494cd7244 100644 --- a/supabase/functions/_backend/utils/stripe.ts +++ b/supabase/functions/_backend/utils/stripe.ts @@ -414,6 +414,7 @@ export async function cancelSubscription(c: Context, customerId: string) { if (!isStripeConfigured(c)) return + let succeeded = true for await (const subscription of getStripe(c).subscriptions.list({ customer: customerId, status: 'all' })) { if (subscription.status === 'canceled' || subscription.status === 'incomplete_expired') continue @@ -422,9 +423,11 @@ export async function cancelSubscription(c: Context, customerId: string) { await getStripe(c).subscriptions.cancel(subscription.id) } catch (error) { + succeeded = false cloudlogErr({ requestId: c.get('requestId'), message: 'cancelSubscription item', error, subscriptionId: subscription.id, customerId }) } } + return succeeded } async function getStoredPlanPriceId(c: Context, planId: string, recurrence: string): Promise { diff --git a/supabase/functions/_backend/utils/supabase.ts b/supabase/functions/_backend/utils/supabase.ts index 3ca4918f17..e7cc017d19 100644 --- a/supabase/functions/_backend/utils/supabase.ts +++ b/supabase/functions/_backend/utils/supabase.ts @@ -1,14 +1,16 @@ import type { SupabaseClient } from '@supabase/supabase-js' import type { Context } from 'hono' -import { HTTPException } from 'hono/http-exception' +// @ts-types="npm:@types/pg" +import type { PoolClient } from 'pg' 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 { HTTPException } from 'hono/http-exception' import { buildBillingPlanBentoTags } from './billing_bento_tags.ts' import { buildNormalizedDeviceForWrite, hasComparableDeviceChanged, nullableString } from './deviceComparison.ts' -import { simpleError, quickError } from './hono.ts' +import { quickError, simpleError } from './hono.ts' import { cloudlog, cloudlogErr } from './logging.ts' import { closeClient, getPgClient } from './pg.ts' import { emptyStatsInsights, normalizeStatsInsightsResult } from './statsInsights.ts' @@ -792,14 +794,21 @@ export async function createApiKey(c: Context, userId: string) { return } - const pgClient = getPgClient(c) + const pgPool = getPgClient(c) + let pgClient: PoolClient | undefined let inTransaction = false try { + pgClient = await pgPool.connect() await pgClient.query('BEGIN') inTransaction = true + await pgClient.query(`SET LOCAL lock_timeout = '5s'`) + // Serialize with delete_user(). Its to_delete_accounts FK check takes a + // KEY SHARE lock on public.users, which conflicts with this UPDATE lock. + // The separate query below therefore gets a fresh READ COMMITTED snapshot + // after any in-flight deletion transaction has finished. const userLockResult = await pgClient.query( - 'SELECT id FROM auth.users WHERE id = $1::uuid FOR UPDATE', + 'SELECT id FROM public.users WHERE id = $1::uuid FOR UPDATE', [userId], ) if (userLockResult.rowCount === 0) { @@ -808,6 +817,20 @@ export async function createApiKey(c: Context, userId: string) { return } + const deletionResult = await pgClient.query<{ deletion_scheduled: boolean }>( + `SELECT EXISTS ( + SELECT 1 + FROM public.to_delete_accounts + WHERE account_id = $1::uuid + ) AS deletion_scheduled`, + [userId], + ) + if (deletionResult.rows[0]?.deletion_scheduled) { + cloudlog({ requestId: c.get('requestId'), message: 'createApiKey skipped, account deletion scheduled', userId }) + await pgClient.query('ROLLBACK') + return + } + const totalResult = await pgClient.query<{ count: string }>( 'SELECT count(*)::text AS count FROM public.apikeys WHERE user_id = $1::uuid', [userId], @@ -963,13 +986,21 @@ export async function createApiKey(c: Context, userId: string) { await pgClient.query('COMMIT') } catch (error) { - if (inTransaction) { + if (inTransaction && pgClient) { await pgClient.query('ROLLBACK').catch(() => {}) } cloudlogErr({ requestId: c.get('requestId'), message: 'createApiKey error', userId, error }) + // A lock timeout is transient: let the queue retry instead of permanently + // acknowledging user provisioning without a default API key. + if (typeof error === 'object' && error !== null && 'code' in error && error.code === '55P03') + throw error } finally { - closeClient(c, pgClient) + // Workerd keeps request-scoped Pools open, so destroy the checked-out + // socket explicitly after the transaction and then close the Pool where + // the runtime supports it. + pgClient?.release(true) + closeClient(c, pgPool) } } diff --git a/supabase/functions/triggers/index.ts b/supabase/functions/triggers/index.ts index 0175d21810..a984057f37 100644 --- a/supabase/functions/triggers/index.ts +++ b/supabase/functions/triggers/index.ts @@ -18,13 +18,14 @@ import { app as on_manifest_create } from '../_backend/triggers/on_manifest_crea import { app as on_org_update } from '../_backend/triggers/on_org_update.ts' import { app as on_organization_create } from '../_backend/triggers/on_organization_create.ts' import { app as on_organization_delete } from '../_backend/triggers/on_organization_delete.ts' -import { app as pluginNotifications } from '../_backend/triggers/plugin_notifications.ts' import { app as on_user_create } from '../_backend/triggers/on_user_create.ts' import { app as on_user_delete } from '../_backend/triggers/on_user_delete.ts' +import { app as on_user_org_access } from '../_backend/triggers/on_user_org_access.ts' import { app as on_user_update } from '../_backend/triggers/on_user_update.ts' import { app as on_version_create } from '../_backend/triggers/on_version_create.ts' import { app as on_version_delete } from '../_backend/triggers/on_version_delete.ts' import { app as on_version_update } from '../_backend/triggers/on_version_update.ts' +import { app as pluginNotifications } from '../_backend/triggers/plugin_notifications.ts' import { app as queue_consumer } from '../_backend/triggers/queue_consumer.ts' import { app as stripe_event } from '../_backend/triggers/stripe_event.ts' import { app as webhook_delivery } from '../_backend/triggers/webhook_delivery.ts' @@ -57,6 +58,7 @@ appGlobal.route('/on_channel_update', on_channel_update) appGlobal.route('/on_user_create', on_user_create) appGlobal.route('/on_user_update', on_user_update) appGlobal.route('/on_user_delete', on_user_delete) +appGlobal.route('/on_user_org_access', on_user_org_access) appGlobal.route('/on_app_create', on_app_create) appGlobal.route('/on_app_delete', on_app_delete) appGlobal.route('/on_app_update', on_app_update) diff --git a/supabase/migrations/20260803161022_on_user_org_access_queue.sql b/supabase/migrations/20260803161022_on_user_org_access_queue.sql new file mode 100644 index 0000000000..0e512232d3 --- /dev/null +++ b/supabase/migrations/20260803161022_on_user_org_access_queue.sql @@ -0,0 +1,90 @@ +-- Feed active direct user-to-organization role bindings into the existing +-- function queue dispatcher. +-- +-- Execution model: +-- - Where: row-level AFTER INSERT / activation-relevant UPDATE trigger on +-- public.role_bindings. +-- - Frequency: once for each qualifying INSERT or activation-relevant UPDATE. +-- Qualifying-to-qualifying updates are deliberate: tag removal is +-- idempotent, the joined event is an at-least-once fact, and a later write +-- can self-heal a previously exhausted delivery. Non-qualifying rows are +-- rejected by the trigger WHEN clause before the trigger function runs. +-- - Roles: any role allowed to write role_bindings; the existing generic +-- trigger function runs as SECURITY DEFINER. +-- - Cardinality: O(1) pgmq.send per qualifying row. The trigger performs no +-- table scan; the queue handler reloads the binding by its primary key. + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pgmq.list_queues() + WHERE queue_name = 'on_user_org_access' + ) THEN + PERFORM pgmq.create('on_user_org_access'); + END IF; +END; +$$; + +DROP TRIGGER IF EXISTS on_user_org_access ON public.role_bindings; + +CREATE TRIGGER on_user_org_access +AFTER INSERT OR UPDATE OF +principal_type, +principal_id, +scope_type, +org_id, +expires_at, +is_direct +ON public.role_bindings +FOR EACH ROW +WHEN ( + new.principal_type = 'user' + AND new.scope_type = 'org' + AND new.org_id IS NOT NULL + AND new.is_direct IS TRUE + AND (new.expires_at IS NULL OR new.expires_at > pg_catalog.now()) +) +EXECUTE FUNCTION public.trigger_http_queue_post_to_function( + 'on_user_org_access' +); + +DO $$ +DECLARE + high_frequency_task_type public.cron_task_type; + high_frequency_target jsonb; +BEGIN + SELECT cron.task_type, cron.target::jsonb + INTO high_frequency_task_type, high_frequency_target + FROM public.cron_tasks AS cron + WHERE cron.name = 'high_frequency_queues' + FOR UPDATE; + + IF NOT FOUND THEN + RAISE EXCEPTION + 'Required cron task high_frequency_queues is missing'; + END IF; + + IF high_frequency_task_type + IS DISTINCT FROM 'function_queue'::public.cron_task_type THEN + RAISE EXCEPTION + 'Cron task high_frequency_queues must use task type function_queue'; + END IF; + + IF pg_catalog.jsonb_typeof(high_frequency_target) + IS DISTINCT FROM 'array' THEN + RAISE EXCEPTION + 'Cron task high_frequency_queues target must be a JSON array'; + END IF; + + IF NOT (high_frequency_target ? 'on_user_org_access') THEN + UPDATE public.cron_tasks + SET + target = ( + high_frequency_target || '["on_user_org_access"]'::jsonb + )::text, + updated_at = pg_catalog.now() + WHERE name = 'high_frequency_queues'; + END IF; +END; +$$; diff --git a/supabase/tests/66_test_on_user_org_access_queue.sql b/supabase/tests/66_test_on_user_org_access_queue.sql new file mode 100644 index 0000000000..cf5a45baf2 --- /dev/null +++ b/supabase/tests/66_test_on_user_org_access_queue.sql @@ -0,0 +1,646 @@ +BEGIN; + +SELECT plan(20); + +CREATE OR REPLACE FUNCTION pg_temp.on_user_org_access_messages( + p_binding_id uuid +) +RETURNS SETOF jsonb +LANGUAGE plpgsql +SET search_path = '' +AS $$ +BEGIN + IF pg_catalog.to_regclass('pgmq.q_on_user_org_access') IS NULL THEN + RETURN; + END IF; + + RETURN QUERY EXECUTE + 'SELECT message + FROM pgmq.q_on_user_org_access + WHERE message -> ''payload'' -> ''record'' ->> ''id'' = $1 + OR message -> ''payload'' -> ''old_record'' ->> ''id'' = $1' + USING p_binding_id::text; +END; +$$; + +CREATE OR REPLACE FUNCTION pg_temp.delete_on_user_org_access_messages( + p_binding_id uuid +) +RETURNS void +LANGUAGE plpgsql +SET search_path = '' +AS $$ +BEGIN + IF pg_catalog.to_regclass('pgmq.q_on_user_org_access') IS NULL THEN + RETURN; + END IF; + + EXECUTE + 'DELETE FROM pgmq.q_on_user_org_access + WHERE message -> ''payload'' -> ''record'' ->> ''id'' = $1 + OR message -> ''payload'' -> ''old_record'' ->> ''id'' = $1' + USING p_binding_id::text; +END; +$$; + +SELECT tests.create_supabase_user( + 'on_user_org_access_actor', + 'on-user-org-access-actor@test.local' +); + +INSERT INTO public.users (id, email, created_at, updated_at) +VALUES ( + tests.get_supabase_uid('on_user_org_access_actor'), + 'on-user-org-access-actor@test.local', + pg_catalog.now(), + pg_catalog.now() +); + +INSERT INTO public.orgs (id, created_by, name, management_email) +VALUES ( + '66000000-0000-4000-8000-000000000100'::uuid, + tests.get_supabase_uid('on_user_org_access_actor'), + 'On User Org Access Queue Test Org', + 'on-user-org-access-org@test.local' +); + +INSERT INTO public.apps (id, app_id, icon_url, user_id, name, owner_org) +VALUES ( + '66000000-0000-4000-8000-000000000200'::uuid, + 'com.test.on-user-org-access-queue', + '', + tests.get_supabase_uid('on_user_org_access_actor'), + 'On User Org Access Queue Test App', + '66000000-0000-4000-8000-000000000100'::uuid +); + +CREATE TEMP TABLE on_user_org_access_context AS +SELECT + '66000000-0000-4000-8000-000000000100'::uuid AS org_id, + '66000000-0000-4000-8000-000000000200'::uuid AS app_id, + tests.get_supabase_uid('on_user_org_access_actor') AS actor_id, + ( + SELECT id + FROM public.roles + WHERE name = public.rbac_role_org_member() + LIMIT 1 + ) AS org_role_id, + ( + SELECT id + FROM public.roles + WHERE name = public.rbac_role_app_developer() + LIMIT 1 + ) AS app_role_id; + +SELECT ok( + EXISTS ( + SELECT 1 + FROM pgmq.list_queues() + WHERE queue_name = 'on_user_org_access' + ), + 'on_user_org_access queue exists' +); + +SELECT is( + ( + SELECT target::jsonb + FROM public.cron_tasks + WHERE name = 'high_frequency_queues' + ), + '[ + "credit_usage_alerts", + "on_app_create", + "on_app_delete", + "on_app_update", + "on_channel_update", + "on_org_update", + "on_organization_create", + "on_user_create", + "on_user_delete", + "on_user_update", + "on_version_create", + "on_version_delete", + "on_version_update", + "webhook_dispatcher", + "webhook_delivery", + "credit_usage_posthog", + "on_user_org_access" + ]'::jsonb, + 'high-frequency queues retain order and append on_user_org_access' +); + +SELECT is( + ( + SELECT count(*) + FROM public.cron_tasks AS cron + CROSS JOIN + LATERAL jsonb_array_elements_text(cron.target::jsonb) + AS queue_name (value) + WHERE + cron.name = 'high_frequency_queues' + AND queue_name.value = 'on_user_org_access' + ), + 1::bigint, + 'high_frequency_queues contains on_user_org_access exactly once' +); + +SELECT is( + ( + SELECT count(*) + FROM public.cron_tasks + WHERE + name = 'high_frequency_queues' + AND task_type = 'function_queue'::public.cron_task_type + AND batch_size = 100 + AND second_interval = 10 + AND minute_interval IS NULL + AND hour_interval IS NULL + AND run_at_hour IS NULL + AND run_at_minute IS NULL + AND run_at_second IS NULL + AND run_on_dow IS NULL + AND run_on_day IS NULL + AND enabled IS TRUE + ), + 1::bigint, + 'high-frequency queues retain task type, batch, interval, and schedule' +); + +SELECT is( + ( + SELECT count(*) + FROM pg_catalog.pg_trigger AS trg + INNER JOIN + pg_catalog.pg_class AS rel + ON trg.tgrelid = rel.oid + INNER JOIN + pg_catalog.pg_namespace AS ns + ON rel.relnamespace = ns.oid + INNER JOIN + pg_catalog.pg_proc AS pgproc + ON trg.tgfoid = pgproc.oid + WHERE + NOT trg.tgisinternal + AND ns.nspname = 'public' + AND rel.relname = 'role_bindings' + AND trg.tgname = 'on_user_org_access' + AND pgproc.proname = 'trigger_http_queue_post_to_function' + ), + 1::bigint, + 'role_bindings has exactly one on_user_org_access generic queue trigger' +); + +SELECT ok( + EXISTS ( + SELECT 1 + FROM pg_catalog.pg_trigger AS trg + INNER JOIN + pg_catalog.pg_class AS rel + ON trg.tgrelid = rel.oid + INNER JOIN + pg_catalog.pg_namespace AS ns + ON rel.relnamespace = ns.oid + WHERE + NOT trg.tgisinternal + AND ns.nspname = 'public' + AND rel.relname = 'role_bindings' + AND trg.tgname = 'on_user_org_access' + AND (trg.tgtype & 1) = 1 + AND (trg.tgtype & 2) = 0 + AND (trg.tgtype & 4) = 4 + AND (trg.tgtype & 8) = 0 + AND (trg.tgtype & 16) = 16 + ), + 'on_user_org_access is row-level AFTER INSERT/UPDATE without DELETE' +); + +SELECT is( + ( + SELECT + pg_catalog.string_agg( + attr.attname, + ',' ORDER BY attr.attname + ) + FROM pg_catalog.pg_trigger AS trg + INNER JOIN + pg_catalog.pg_class AS rel + ON trg.tgrelid = rel.oid + INNER JOIN + pg_catalog.pg_namespace AS ns + ON rel.relnamespace = ns.oid + CROSS JOIN + LATERAL pg_catalog.unnest(trg.tgattr::smallint []) + AS update_column (attnum) + INNER JOIN pg_catalog.pg_attribute AS attr + ON + rel.oid = attr.attrelid + AND update_column.attnum = attr.attnum + WHERE + NOT trg.tgisinternal + AND ns.nspname = 'public' + AND rel.relname = 'role_bindings' + AND trg.tgname = 'on_user_org_access' + ), + 'expires_at,is_direct,org_id,principal_id,principal_type,scope_type', + 'UPDATE events are limited to activation-relevant columns' +); + +SELECT ok( + EXISTS ( + SELECT 1 + FROM pg_catalog.pg_trigger AS trg + INNER JOIN + pg_catalog.pg_class AS rel + ON trg.tgrelid = rel.oid + INNER JOIN + pg_catalog.pg_namespace AS ns + ON rel.relnamespace = ns.oid + CROSS JOIN + LATERAL ( + SELECT pg_catalog.pg_get_triggerdef(trg.oid) AS definition + ) AS predicate + WHERE + NOT trg.tgisinternal + AND ns.nspname = 'public' + AND rel.relname = 'role_bindings' + AND trg.tgname = 'on_user_org_access' + AND predicate.definition LIKE '%principal_type%user%' + AND predicate.definition LIKE '%scope_type%org%' + AND predicate.definition LIKE '%org_id%IS NOT NULL%' + AND predicate.definition LIKE '%is_direct%' + AND predicate.definition LIKE '%expires_at%IS NULL%' + AND predicate.definition LIKE '%expires_at%now()%' + ), + 'trigger has an active direct user-to-organization WHEN predicate' +); + +INSERT INTO public.role_bindings ( + id, + principal_type, + principal_id, + role_id, + scope_type, + org_id, + granted_by, + reason, + is_direct +) +SELECT + '66000000-0000-4000-8000-000000000001'::uuid AS id, + public.rbac_principal_user() AS principal_type, + '66000000-0000-4000-8000-000000001001'::uuid AS principal_id, + org_role_id AS role_id, + public.rbac_scope_org() AS scope_type, + org_id, + actor_id AS granted_by, + 'qualifying active direct binding' AS reason, + TRUE AS is_direct +FROM on_user_org_access_context; + +SELECT is( + ( + SELECT message + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000001'::uuid + ) AS message + ), + ( + SELECT + pg_catalog.jsonb_build_object( + 'function_name', 'on_user_org_access', + 'function_type', 'cloudflare', + 'payload', pg_catalog.jsonb_build_object( + 'old_record', 'null'::jsonb, + 'record', pg_catalog.to_jsonb(rb), -- noqa: RF03 + 'type', 'INSERT', + 'table', 'role_bindings', + 'schema', 'public' + ) + ) + FROM public.role_bindings AS rb + WHERE rb.id = '66000000-0000-4000-8000-000000000001'::uuid -- noqa: RF03 + ), + 'active direct user/org INSERT queues the exact generic envelope' +); + +INSERT INTO public.role_bindings ( + id, + principal_type, + principal_id, + role_id, + scope_type, + org_id, + granted_by, + expires_at, + reason, + is_direct +) +SELECT + '66000000-0000-4000-8000-000000000002'::uuid AS id, + public.rbac_principal_user() AS principal_type, + '66000000-0000-4000-8000-000000001002'::uuid AS principal_id, + org_role_id AS role_id, + public.rbac_scope_org() AS scope_type, + org_id, + actor_id AS granted_by, + pg_catalog.now() - interval '1 hour' AS expires_at, + 'expired binding' AS reason, + TRUE AS is_direct +FROM on_user_org_access_context; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000002'::uuid + ) + ), + 0::bigint, + 'expired user-to-organization INSERT does not queue' +); + +INSERT INTO public.role_bindings ( + id, + principal_type, + principal_id, + role_id, + scope_type, + org_id, + granted_by, + reason, + is_direct +) +SELECT + '66000000-0000-4000-8000-000000000003'::uuid AS id, + public.rbac_principal_apikey() AS principal_type, + '66000000-0000-4000-8000-000000001003'::uuid AS principal_id, + org_role_id AS role_id, + public.rbac_scope_org() AS scope_type, + org_id, + actor_id AS granted_by, + 'API key binding' AS reason, + TRUE AS is_direct +FROM on_user_org_access_context; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000003'::uuid + ) + ), + 0::bigint, + 'API key organization binding INSERT does not queue' +); + +INSERT INTO public.role_bindings ( + id, + principal_type, + principal_id, + role_id, + scope_type, + org_id, + app_id, + granted_by, + reason, + is_direct +) +SELECT + '66000000-0000-4000-8000-000000000004'::uuid AS id, + public.rbac_principal_user() AS principal_type, + '66000000-0000-4000-8000-000000001004'::uuid AS principal_id, + app_role_id AS role_id, + public.rbac_scope_app() AS scope_type, + org_id, + app_id, + actor_id AS granted_by, + 'app-scoped binding' AS reason, + TRUE AS is_direct +FROM on_user_org_access_context; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000004'::uuid + ) + ), + 0::bigint, + 'app-scoped user binding INSERT does not queue' +); + +SELECT throws_ok( + $q$ + INSERT INTO public.role_bindings ( + id, + principal_type, + principal_id, + role_id, + scope_type, + org_id, + granted_by, + reason, + is_direct + ) + SELECT + '66000000-0000-4000-8000-000000000005'::uuid, + public.rbac_principal_user(), + '66000000-0000-4000-8000-000000001005'::uuid, + org_role_id, + public.rbac_scope_org(), + NULL, + actor_id, + 'invalid null-org binding', + true + FROM on_user_org_access_context; + $q$, + '23514', + pg_catalog.concat( + 'new row for relation "role_bindings" violates check constraint ', + '"role_bindings_check"' + ), + 'organization-scoped binding with a null organization is rejected' +); + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000005'::uuid + ) + ), + 0::bigint, + 'rejected null-organization INSERT does not queue' +); + +INSERT INTO public.role_bindings ( + id, + principal_type, + principal_id, + role_id, + scope_type, + org_id, + granted_by, + reason, + is_direct +) +SELECT + '66000000-0000-4000-8000-000000000006'::uuid AS id, + public.rbac_principal_user() AS principal_type, + '66000000-0000-4000-8000-000000001006'::uuid AS principal_id, + org_role_id AS role_id, + public.rbac_scope_org() AS scope_type, + org_id, + actor_id AS granted_by, + 'indirect binding' AS reason, + FALSE AS is_direct +FROM on_user_org_access_context; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000006'::uuid + ) + ), + 0::bigint, + 'indirect user-to-organization INSERT does not queue' +); + +CREATE TEMP TABLE on_user_org_access_old_records AS +SELECT + rb.id AS binding_id, + pg_catalog.to_jsonb(rb) AS record -- noqa: RF03 +FROM public.role_bindings AS rb +WHERE rb.id IN ( + '66000000-0000-4000-8000-000000000002'::uuid, + '66000000-0000-4000-8000-000000000006'::uuid +); + +UPDATE public.role_bindings +SET expires_at = NULL +WHERE id = '66000000-0000-4000-8000-000000000002'::uuid; + +SELECT is( + ( + SELECT message + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000002'::uuid + ) AS message + ), + ( + SELECT + pg_catalog.jsonb_build_object( + 'function_name', 'on_user_org_access', + 'function_type', 'cloudflare', + 'payload', pg_catalog.jsonb_build_object( + 'old_record', old_record.record, + 'record', pg_catalog.to_jsonb(rb), -- noqa: RF02 + 'type', 'UPDATE', + 'table', 'role_bindings', + 'schema', 'public' + ) + ) + FROM public.role_bindings AS rb + INNER JOIN on_user_org_access_old_records AS old_record + ON rb.id = old_record.binding_id + WHERE rb.id = '66000000-0000-4000-8000-000000000002'::uuid + ), + 'expired-to-active UPDATE queues the exact generic old/new envelope' +); + +DO $$ +BEGIN + PERFORM pg_temp.delete_on_user_org_access_messages( + '66000000-0000-4000-8000-000000000002'::uuid + ); +END; +$$; + +UPDATE public.role_bindings +SET expires_at = pg_catalog.now() + interval '1 day' +WHERE id = '66000000-0000-4000-8000-000000000002'::uuid; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000002'::uuid + ) + ), + 1::bigint, + 'active-to-active UPDATE queues again for self-healing delivery' +); + +UPDATE public.role_bindings +SET is_direct = TRUE +WHERE id = '66000000-0000-4000-8000-000000000006'::uuid; + +SELECT is( + ( + SELECT message + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000006'::uuid + ) AS message + ), + ( + SELECT + pg_catalog.jsonb_build_object( + 'function_name', 'on_user_org_access', + 'function_type', 'cloudflare', + 'payload', pg_catalog.jsonb_build_object( + 'old_record', old_record.record, + 'record', pg_catalog.to_jsonb(rb), -- noqa: RF02 + 'type', 'UPDATE', + 'table', 'role_bindings', + 'schema', 'public' + ) + ) + FROM public.role_bindings AS rb + INNER JOIN on_user_org_access_old_records AS old_record + ON rb.id = old_record.binding_id + WHERE rb.id = '66000000-0000-4000-8000-000000000006'::uuid + ), + 'indirect-to-direct UPDATE queues the exact generic old/new envelope' +); + +DO $$ +BEGIN + PERFORM pg_temp.delete_on_user_org_access_messages( + '66000000-0000-4000-8000-000000000001'::uuid + ); +END; +$$; + +UPDATE public.role_bindings +SET reason = 'reason-only update must not queue' +WHERE id = '66000000-0000-4000-8000-000000000001'::uuid; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000001'::uuid + ) + ), + 0::bigint, + 'reason-only UPDATE does not queue' +); + +DELETE FROM public.role_bindings +WHERE id = '66000000-0000-4000-8000-000000000001'::uuid; + +SELECT is( + ( + SELECT count(*) + FROM pg_temp.on_user_org_access_messages( + '66000000-0000-4000-8000-000000000001'::uuid + ) + ), + 0::bigint, + 'DELETE does not queue' +); + +SELECT * FROM finish(); -- noqa: AM04 + +ROLLBACK; diff --git a/tests/bento-abort-signal.unit.test.ts b/tests/bento-abort-signal.unit.test.ts new file mode 100644 index 0000000000..e8e65291c0 --- /dev/null +++ b/tests/bento-abort-signal.unit.test.ts @@ -0,0 +1,106 @@ +import type { Context } from 'hono' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { syncBentoSubscriberTags, unsubscribeBento } from '../supabase/functions/_backend/utils/bento.ts' + +vi.mock('../supabase/functions/_backend/utils/logging.ts', () => ({ + cloudlog: vi.fn(), + cloudlogErr: vi.fn(), + serializeError: (error: unknown) => error, +})) + +vi.mock('../supabase/functions/_backend/utils/utils.ts', () => ({ + getEnv: (_context: unknown, key: string) => { + const values: Record = { + BENTO_PUBLISHABLE_KEY: 'publishable-key-value', + BENTO_SECRET_KEY: 'secret-key-value', + BENTO_SITE_UUID: 'site-uuid-value', + } + return values[key] ?? '' + }, +})) + +function createContext() { + return { + get: vi.fn(() => 'request-id'), + } as unknown as Context +} + +const subscriberUpdate = { + deleteSegments: ['onboarding:awaiting_first_org'], + email: 'deleted.user@example.com', + segments: ['onboarding:first_org_recovery_suppressed'], +} + +describe('bento abort signals', () => { + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('forwards each optional signal to the exact Bento fetch', async () => { + const fetchMock = vi.fn<( + input: string | URL | Request, + init?: RequestInit, + ) => Promise>(async () => new Response(JSON.stringify({ + failed: 0, + results: 1, + }), { + headers: { 'content-type': 'application/json' }, + status: 200, + })) + vi.stubGlobal('fetch', fetchMock) + + const subscriberController = new AbortController() + const unsubscribeController = new AbortController() + await expect(syncBentoSubscriberTags( + createContext(), + subscriberUpdate, + subscriberController.signal, + )).resolves.toBe(true) + await expect(unsubscribeBento( + createContext(), + subscriberUpdate.email, + unsubscribeController.signal, + )).resolves.toBe(true) + + expect(fetchMock).toHaveBeenCalledTimes(2) + const [subscriberUrl, subscriberInit] = fetchMock.mock.calls[0]! + const [unsubscribeUrl, unsubscribeInit] = fetchMock.mock.calls[1]! + expect(String(subscriberUrl)).toContain('/api/v1/batch/subscribers') + expect(subscriberInit?.signal).toBe(subscriberController.signal) + expect(String(unsubscribeUrl)).toContain('/api/v1/fetch/commands') + expect(unsubscribeInit?.signal).toBe(unsubscribeController.signal) + }) + + it.each([ + [ + 'subscriber synchronization', + (context: Context, signal: AbortSignal) => syncBentoSubscriberTags(context, subscriberUpdate, signal), + ], + [ + 'unsubscribe', + (context: Context, signal: AbortSignal) => unsubscribeBento(context, subscriberUpdate.email, signal), + ], + ])('settles %s as failed when its fetch is aborted', async (_label, operation) => { + const fetchMock = vi.fn((_input: string | URL | Request, init?: RequestInit) => new Promise((_resolve, reject) => { + const signal = init?.signal + if (!signal) { + reject(new Error('missing abort signal')) + return + } + const rejectAbort = () => reject(new DOMException('The operation was aborted', 'AbortError')) + if (signal.aborted) + rejectAbort() + else + signal.addEventListener('abort', rejectAbort, { once: true }) + })) + vi.stubGlobal('fetch', fetchMock) + + const controller = new AbortController() + const result = operation(createContext(), controller.signal) + expect(fetchMock).toHaveBeenCalledOnce() + controller.abort() + + await expect(result).resolves.toBe(false) + }) +}) diff --git a/tests/bento-first-org-lifecycle.unit.test.ts b/tests/bento-first-org-lifecycle.unit.test.ts new file mode 100644 index 0000000000..963858b49d --- /dev/null +++ b/tests/bento-first-org-lifecycle.unit.test.ts @@ -0,0 +1,946 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { + closeClientMock, + createApiKeyMock, + getPgClientMock, + pgConnectMock, + pgQueryMock, + pgReleaseMock, + sendEventToTrackingMock, + syncBentoSubscriberTagsMock, + syncUserPreferenceTagsMock, + trackBentoEventMock, + unsubscribeBentoMock, +} = vi.hoisted(() => { + const pgQueryMock = vi.fn<(query: string, params?: unknown[]) => Promise<{ rows: Record[] }>>(async () => ({ rows: [] })) + const pgReleaseMock = vi.fn<(destroy?: Error | boolean) => void>(() => undefined) + const pgConnectMock = vi.fn(async () => ({ query: pgQueryMock, release: pgReleaseMock })) + return { + closeClientMock: vi.fn(async () => undefined), + createApiKeyMock: vi.fn(async () => undefined), + getPgClientMock: vi.fn(() => ({ connect: pgConnectMock, query: pgQueryMock })), + pgConnectMock, + pgQueryMock, + pgReleaseMock, + sendEventToTrackingMock: vi.fn(async () => undefined), + syncBentoSubscriberTagsMock: vi.fn<( + c: unknown, + update: { deleteSegments: string[], email: string, segments: string[] } + | Array<{ deleteSegments: string[], email: string, segments: string[] }>, + signal?: AbortSignal, + ) => Promise>(async () => true), + syncUserPreferenceTagsMock: vi.fn(async () => undefined), + trackBentoEventMock: vi.fn(async () => true as boolean | undefined), + unsubscribeBentoMock: vi.fn<( + c: unknown, + email: string, + signal?: AbortSignal, + ) => Promise>(async () => true), + } +}) + +vi.mock('../supabase/functions/_backend/utils/bento.ts', () => ({ + syncBentoSubscriberTags: syncBentoSubscriberTagsMock, + trackBentoEvent: trackBentoEventMock, + unsubscribeBento: unsubscribeBentoMock, +})) + +vi.mock('../supabase/functions/_backend/utils/hono.ts', async () => { + const actual = await vi.importActual('../supabase/functions/_backend/utils/hono.ts') + return { + ...actual, + middlewareAPISecret: async (_c: unknown, next: () => Promise) => await next(), + } +}) + +vi.mock('../supabase/functions/_backend/utils/pg.ts', () => ({ + closeClient: closeClientMock, + getPgClient: getPgClientMock, +})) + +vi.mock('../supabase/functions/_backend/utils/supabase.ts', () => ({ + createApiKey: createApiKeyMock, +})) + +vi.mock('../supabase/functions/_backend/utils/tracking.ts', () => ({ + sendEventToTracking: sendEventToTrackingMock, +})) + +vi.mock('../supabase/functions/_backend/utils/user_preferences.ts', () => ({ + syncUserPreferenceTags: syncUserPreferenceTagsMock, +})) + +const { app: onUserCreateApp } = await import('../supabase/functions/_backend/triggers/on_user_create.ts') +const bentoFirstOrgLifecycle = await import('../supabase/functions/_backend/utils/bento_first_org.ts') + +const USER_ID = '11111111-1111-4111-8111-111111111111' +const REGISTERED_AT = '2026-08-03T08:30:00.000Z' +const JOINED_AT = '2026-08-03T09:15:00.000Z' +const LIFECYCLE_TAG = 'onboarding:awaiting_first_org' +const SUPPRESSION_TAG = 'onboarding:first_org_recovery_suppressed' +const ORG_ID = '22222222-2222-4222-8222-222222222222' +const ROLE_BINDING_ID = '33333333-3333-4333-8333-333333333333' + +type ExpectedLifecycleModule = typeof bentoFirstOrgLifecycle & { + syncBentoFirstOrgOnEmailChange: (c: never, oldEmail: string, newEmail: string) => Promise + syncBentoFirstOrgOnRoleBindingWrite: (c: never, roleBindingId: string) => Promise +} + +async function suppressEmailAliases(oldEmail: string, newEmail: string) { + return await (bentoFirstOrgLifecycle as ExpectedLifecycleModule) + .syncBentoFirstOrgOnEmailChange({ get: vi.fn(() => 'request-id') } as never, oldEmail, newEmail) +} + +async function syncRoleBinding(roleBindingId = ROLE_BINDING_ID) { + return await (bentoFirstOrgLifecycle as ExpectedLifecycleModule) + .syncBentoFirstOrgOnRoleBindingWrite({ get: vi.fn(() => 'request-id') } as never, roleBindingId) +} + +function activeBinding(overrides: Record = {}) { + return { + email: ' Current.User@Example.COM ', + granted_at: new Date(JOINED_AT), + id: ROLE_BINDING_ID, + is_active: true, + is_direct: true, + org_id: ORG_ID, + principal_id: USER_ID, + principal_type: 'user', + scope_type: 'org', + ...overrides, + } +} + +function firstOrgDatabaseState(hasActiveDirectOrgAccess = false, userIsRecoveryEligible = true) { + return { + has_active_direct_org_access: hasActiveDirectOrgAccess, + user_is_recovery_eligible: userIsRecoveryEligible, + } +} + +function userRecord(overrides: Record = {}) { + return { + ban_time: null, + country: null, + created_at: REGISTERED_AT, + created_via_invite: false, + discord_username: null, + email: ' New.User@Example.COM ', + email_preferences: {}, + enable_notifications: false, + first_name: 'New', + format_locale: null, + github_id: null, + github_username: null, + id: USER_ID, + image_url: null, + last_name: 'User', + opt_for_newsletters: false, + updated_at: REGISTERED_AT, + ...overrides, + } +} + +async function postUser(record = userRecord()) { + return await onUserCreateApp.request('http://local/', { + body: JSON.stringify({ + old_record: null, + record, + schema: 'public', + table: 'users', + type: 'INSERT', + }), + headers: { 'content-type': 'application/json' }, + method: 'POST', + }) +} + +describe('first-organization lifecycle on user registration', () => { + beforeEach(() => { + vi.clearAllMocks() + pgConnectMock.mockImplementation(async () => ({ query: pgQueryMock, release: pgReleaseMock })) + pgReleaseMock.mockImplementation(() => undefined) + closeClientMock.mockResolvedValue(undefined) + getPgClientMock.mockImplementation(() => ({ connect: pgConnectMock, query: pgQueryMock })) + pgQueryMock.mockResolvedValue({ rows: [firstOrgDatabaseState()] }) + syncBentoSubscriberTagsMock.mockResolvedValue(true) + trackBentoEventMock.mockResolvedValue(true) + unsubscribeBentoMock.mockResolvedValue(true) + }) + + afterEach(() => { + vi.restoreAllMocks() + }) + + it('adds the lifecycle tag and emits the entry event when no active org access exists', async () => { + const response = await postUser() + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), { + deleteSegments: [], + email: 'new.user@example.com', + segments: [LIFECYCLE_TAG], + }) + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'new.user@example.com', + expect.objectContaining({ id: USER_ID }), + ) + expect(trackBentoEventMock).toHaveBeenCalledWith( + expect.anything(), + 'new.user@example.com', + { + created_via_invite: false, + registered_at: REGISTERED_AT, + user_id: USER_ID, + }, + 'user:registered_without_org', + ) + expect(pgQueryMock).toHaveBeenCalledTimes(4) + }) + + it('returns a retryable failure when default API-key provisioning times out on a lock', async () => { + createApiKeyMock.mockRejectedValueOnce(Object.assign(new Error('lock timeout'), { code: '55P03' })) + + const response = await postUser() + + expect(response.status).toBe(500) + expect(createApiKeyMock).toHaveBeenCalledOnce() + expect(syncUserPreferenceTagsMock).not.toHaveBeenCalled() + expect(syncBentoSubscriberTagsMock).not.toHaveBeenCalled() + expect(trackBentoEventMock).not.toHaveBeenCalled() + }) + + it('destroys each checked-out Workerd client before the following Bento request', async () => { + const lifecycleTrace: string[] = [] + let queryNumber = 0 + pgQueryMock.mockImplementation(async () => { + lifecycleTrace.push(`query:${++queryNumber}`) + return { rows: [firstOrgDatabaseState()] } + }) + // Workerd closeClient is intentionally a no-op. This test must prove the + // checked-out clients are released without relying on pool shutdown. + closeClientMock.mockImplementation(async () => undefined) + pgReleaseMock.mockImplementation((destroy) => { + lifecycleTrace.push(`client:released:${destroy}`) + }) + syncBentoSubscriberTagsMock.mockImplementation(async () => { + lifecycleTrace.push('tag:add') + return true + }) + trackBentoEventMock.mockImplementation(async () => { + lifecycleTrace.push('event:entry') + return true + }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(lifecycleTrace).toEqual([ + 'query:1', + 'client:released:true', + 'query:2', + 'client:released:true', + 'tag:add', + 'query:3', + 'client:released:true', + 'event:entry', + 'query:4', + 'client:released:true', + ]) + expect(getPgClientMock).toHaveBeenCalledTimes(2) + expect(pgConnectMock).toHaveBeenCalledTimes(4) + expect(pgReleaseMock).toHaveBeenCalledTimes(4) + expect(pgReleaseMock).toHaveBeenNthCalledWith(1, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(2, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(3, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(4, true) + expect(closeClientMock).toHaveBeenCalledTimes(2) + }) + + it('skips entry and permanently suppresses invite-created profiles', async () => { + const response = await postUser(userRecord({ created_via_invite: true })) + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(pgQueryMock).toHaveBeenCalledTimes(3) + }) + + it.each([ + ['invite-created profile', userRecord({ created_via_invite: true }), firstOrgDatabaseState()], + ['existing organization member', userRecord(), firstOrgDatabaseState(true)], + ])('suppresses a %s when deletion appears after terminal tag removal', async (_label, record, branchState) => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [branchState] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + const response = await postUser(record) + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(1, expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + }) + + it('suppresses an invite-created profile when deletion is scheduled during provisioning', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + const response = await postUser(userRecord({ created_via_invite: true })) + + expect(response.status).toBe(200) + expect(createApiKeyMock).toHaveBeenCalledOnce() + expect(syncUserPreferenceTagsMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + expect(trackBentoEventMock).not.toHaveBeenCalled() + }) + + it('permanently suppresses recovery when a delayed create message finds deletion scheduled', async () => { + pgQueryMock.mockResolvedValue({ rows: [firstOrgDatabaseState(false, false)] }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(pgQueryMock).toHaveBeenCalledOnce() + expect(createApiKeyMock).not.toHaveBeenCalled() + expect(syncUserPreferenceTagsMock).not.toHaveBeenCalled() + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + expect(syncBentoSubscriberTagsMock.mock.invocationCallOrder[0]) + .toBeLessThan(unsubscribeBentoMock.mock.invocationCallOrder[0]) + }) + + it('still attempts the final unsubscribe when deletion suppression throws', async () => { + pgQueryMock.mockResolvedValue({ rows: [firstOrgDatabaseState(false, false)] }) + syncBentoSubscriberTagsMock.mockRejectedValueOnce(new Error('Bento unavailable')) + + const response = await postUser() + + expect(response.status).toBe(500) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + expect(createApiKeyMock).not.toHaveBeenCalled() + expect(syncUserPreferenceTagsMock).not.toHaveBeenCalled() + }) + + it('skips entry and permanently suppresses a user with active direct org access', async () => { + pgQueryMock.mockResolvedValue({ rows: [firstOrgDatabaseState(true)] }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(pgQueryMock).toHaveBeenCalledTimes(3) + }) + + it('suppresses an existing organization member when deletion is scheduled during provisioning', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(true)] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(createApiKeyMock).toHaveBeenCalledOnce() + expect(syncUserPreferenceTagsMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + expect(trackBentoEventMock).not.toHaveBeenCalled() + }) + + it('restricts the active-access check to active direct user-to-org bindings', async () => { + const response = await postUser() + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), { + deleteSegments: [], + email: 'new.user@example.com', + segments: [LIFECYCLE_TAG], + }) + expect(trackBentoEventMock).toHaveBeenCalledTimes(1) + + const normalizedQuery = String(pgQueryMock.mock.calls[0]?.[0]).replace(/\s+/g, ' ') + expect(normalizedQuery).toContain('principal_type = \'user\'') + expect(normalizedQuery).toContain('FROM public.users AS users') + expect(normalizedQuery).toContain('FROM public.to_delete_accounts AS deleted') + expect(normalizedQuery).toContain('scope_type = \'org\'') + expect(normalizedQuery).toContain('is_direct IS TRUE') + expect(normalizedQuery).toContain('org_id IS NOT NULL') + expect(normalizedQuery).toContain('(expires_at IS NULL OR expires_at > pg_catalog.now())') + expect(pgQueryMock.mock.calls[0]?.[1]).toEqual([USER_ID]) + }) + + it('permanently suppresses recovery when org access appears after the tag write', async () => { + const lifecycleTrace: string[] = [] + pgQueryMock + .mockImplementationOnce(async () => { + lifecycleTrace.push('query:preflight-no-access') + return { rows: [firstOrgDatabaseState()] } + }) + .mockImplementationOnce(async () => { + lifecycleTrace.push('query:post-provision-no-access') + return { rows: [firstOrgDatabaseState()] } + }) + .mockImplementationOnce(async () => { + lifecycleTrace.push('query:access-found') + return { rows: [firstOrgDatabaseState(true)] } + }) + .mockImplementationOnce(async () => { + lifecycleTrace.push('query:post-remove') + return { rows: [firstOrgDatabaseState(true)] } + }) + syncBentoSubscriberTagsMock.mockImplementation(async (_c, update) => { + await Promise.resolve() + const updates = Array.isArray(update) ? update : [update] + if (updates.some(item => item.segments.includes(LIFECYCLE_TAG))) + lifecycleTrace.push('tag:add') + if (updates.some(item => item.deleteSegments.includes(LIFECYCLE_TAG))) + lifecycleTrace.push('tag:suppress') + return true + }) + // Model Workerd: pool close does nothing, so only explicit client release + // can appear before the external Bento operations. + closeClientMock.mockImplementation(async () => undefined) + pgReleaseMock.mockImplementation((destroy) => { + lifecycleTrace.push(`client:released:${destroy}`) + }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(lifecycleTrace).toEqual([ + 'query:preflight-no-access', + 'client:released:true', + 'query:post-provision-no-access', + 'client:released:true', + 'tag:add', + 'query:access-found', + 'client:released:true', + 'tag:suppress', + 'query:post-remove', + 'client:released:true', + ]) + expect(getPgClientMock).toHaveBeenCalledTimes(2) + expect(pgConnectMock).toHaveBeenCalledTimes(4) + expect(pgReleaseMock).toHaveBeenNthCalledWith(1, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(2, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(3, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(4, true) + expect(closeClientMock).toHaveBeenCalledTimes(2) + expect(pgQueryMock).toHaveBeenCalledTimes(4) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledTimes(2) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(1, expect.anything(), { + deleteSegments: [], + email: 'new.user@example.com', + segments: [LIFECYCLE_TAG], + }) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(trackBentoEventMock).not.toHaveBeenCalled() + }) + + it('suppresses recovery when deletion is scheduled after the tag write', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(1, expect.anything(), { + deleteSegments: [], + email: 'new.user@example.com', + segments: [LIFECYCLE_TAG], + }) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(createApiKeyMock).toHaveBeenCalledOnce() + expect(syncUserPreferenceTagsMock).toHaveBeenCalledOnce() + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + }) + + it('suppresses and unsubscribes when deletion is scheduled after the entry event', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(trackBentoEventMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(trackBentoEventMock.mock.invocationCallOrder[0]) + .toBeLessThan(syncBentoSubscriberTagsMock.mock.invocationCallOrder[1]!) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + }) + + it('permanently suppresses recovery when org access appears after the entry event', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(true)] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(true)] }) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(trackBentoEventMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(unsubscribeBentoMock).not.toHaveBeenCalled() + }) + + it.each([ + ['returns false', async () => false], + ['throws', async () => { throw new Error('Bento unavailable') }], + ])('fails the handler for retry when configured Bento %s', async (_label, failure) => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + syncBentoSubscriberTagsMock.mockImplementationOnce(failure) + + const response = await postUser() + + expect(response.status).toBe(500) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + }) + + it('fails the handler for retry when the entry event returns false', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + trackBentoEventMock.mockResolvedValue(false) + + const response = await postUser() + + expect(response.status).toBe(500) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'new.user@example.com', expect.any(AbortSignal)) + }) + + it('succeeds as a no-op when Bento is not configured', async () => { + syncBentoSubscriberTagsMock.mockResolvedValue(undefined) + trackBentoEventMock.mockResolvedValue(undefined) + + const response = await postUser() + + expect(response.status).toBe(200) + expect(trackBentoEventMock).toHaveBeenCalledTimes(1) + }) +}) + +describe('first-organization recovery suppression on email changes', () => { + beforeEach(() => { + vi.clearAllMocks() + pgConnectMock.mockImplementation(async () => ({ query: pgQueryMock, release: pgReleaseMock })) + pgReleaseMock.mockImplementation(() => undefined) + closeClientMock.mockResolvedValue(undefined) + getPgClientMock.mockImplementation(() => ({ connect: pgConnectMock, query: pgQueryMock })) + pgQueryMock.mockResolvedValue({ rows: [firstOrgDatabaseState()] }) + syncBentoSubscriberTagsMock.mockResolvedValue(true) + trackBentoEventMock.mockResolvedValue(true) + }) + + afterEach(() => { + vi.restoreAllMocks() + }) + + it('suppresses both unique normalized aliases and clears awaiting state in one batch', async () => { + await expect(suppressEmailAliases(' Old.User@Example.COM ', ' New.User@Example.COM ')).resolves.toBe(true) + + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [ + { + deleteSegments: [LIFECYCLE_TAG], + email: 'old.user@example.com', + segments: [SUPPRESSION_TAG], + }, + { + deleteSegments: [LIFECYCLE_TAG], + email: 'new.user@example.com', + segments: [SUPPRESSION_TAG], + }, + ]) + }) + + it('deduplicates the same normalized alias within a batch', async () => { + await expect(suppressEmailAliases(' Same.User@Example.COM ', 'same.user@example.com')).resolves.toBe(true) + + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'same.user@example.com', + segments: [SUPPRESSION_TAG], + }]) + }) + + it('returns false for configured delivery failure and undefined when unconfigured', async () => { + syncBentoSubscriberTagsMock.mockResolvedValueOnce(false) + await expect(suppressEmailAliases('a@example.com', 'b@example.com')).resolves.toBe(false) + + syncBentoSubscriberTagsMock.mockResolvedValueOnce(undefined) + await expect(suppressEmailAliases('a@example.com', 'b@example.com')).resolves.toBeUndefined() + }) + + it('keeps suppression monotonic across reversed, repeated, and chained alias changes', async () => { + await suppressEmailAliases('A@Example.com', 'b@example.com') + await suppressEmailAliases('b@example.com', 'A@example.com') + await suppressEmailAliases('a@example.com', 'b@example.com') + await suppressEmailAliases('b@example.com', 'c@example.com') + await suppressEmailAliases('C@example.com', 'c@example.com') + + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledTimes(5) + const expectedAliasBatches = [ + ['a@example.com', 'b@example.com'], + ['b@example.com', 'a@example.com'], + ['a@example.com', 'b@example.com'], + ['b@example.com', 'c@example.com'], + ['c@example.com'], + ] + const requestedAliases = new Set() + for (const [index, [, rawUpdate]] of syncBentoSubscriberTagsMock.mock.calls.entries()) { + const updates = Array.isArray(rawUpdate) ? rawUpdate : [rawUpdate] + expect(updates.map(update => update.email)).toEqual(expectedAliasBatches[index]) + expect(new Set(updates.map(update => update.email)).size).toBe(updates.length) + for (const update of updates) { + requestedAliases.add(update.email) + expect(update.segments).toEqual([SUPPRESSION_TAG]) + expect(update.deleteSegments).toEqual([LIFECYCLE_TAG]) + expect(update.deleteSegments).not.toContain(SUPPRESSION_TAG) + } + } + expect(requestedAliases).toEqual(new Set(['a@example.com', 'b@example.com', 'c@example.com'])) + }) + + it('never removes permanent suppression when signup later adds awaiting state', async () => { + await suppressEmailAliases('old@example.com', 'new@example.com') + const response = await postUser(userRecord({ email: 'new@example.com' })) + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledTimes(2) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), { + deleteSegments: [], + email: 'new@example.com', + segments: [LIFECYCLE_TAG], + }) + for (const [, rawUpdate] of syncBentoSubscriberTagsMock.mock.calls) { + const updates = Array.isArray(rawUpdate) ? rawUpdate : [rawUpdate] + for (const update of updates) + expect(update.deleteSegments).not.toContain(SUPPRESSION_TAG) + } + }) +}) + +describe('first-organization lifecycle on direct org access', () => { + beforeEach(() => { + vi.clearAllMocks() + pgConnectMock.mockImplementation(async () => ({ query: pgQueryMock, release: pgReleaseMock })) + pgReleaseMock.mockImplementation(() => undefined) + closeClientMock.mockResolvedValue(undefined) + getPgClientMock.mockImplementation(() => ({ connect: pgConnectMock, query: pgQueryMock })) + pgQueryMock.mockImplementation(async query => ({ + rows: String(query).includes('FROM public.role_bindings AS binding') + ? [activeBinding()] + : [firstOrgDatabaseState()], + })) + syncBentoSubscriberTagsMock.mockResolvedValue(true) + trackBentoEventMock.mockResolvedValue(true) + }) + + afterEach(() => { + vi.restoreAllMocks() + }) + + it('atomically suppresses recovery and emits a joined-org fact for a qualifying binding', async () => { + await syncRoleBinding() + + expect(pgQueryMock).toHaveBeenCalledWith(expect.stringContaining('FROM public.role_bindings'), [ROLE_BINDING_ID]) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'current.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(trackBentoEventMock).toHaveBeenCalledWith( + expect.anything(), + 'current.user@example.com', + { + joined_at: JOINED_AT, + org_id: ORG_ID, + role_binding_id: ROLE_BINDING_ID, + user_id: USER_ID, + }, + 'user:joined_org', + ) + }) + + it('suppresses and unsubscribes when deletion is scheduled during joined-org delivery', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [activeBinding()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(true)] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + await syncRoleBinding() + + expect(trackBentoEventMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'current.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(trackBentoEventMock.mock.invocationCallOrder[0]) + .toBeLessThan(syncBentoSubscriberTagsMock.mock.invocationCallOrder[1]!) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'current.user@example.com', expect.any(AbortSignal)) + }) + + it('suppresses without emitting joined-org state when deletion is already scheduled', async () => { + pgQueryMock + .mockResolvedValueOnce({ rows: [activeBinding()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + + await syncRoleBinding() + + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'current.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(trackBentoEventMock).not.toHaveBeenCalled() + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'current.user@example.com', expect.any(AbortSignal)) + }) + + it('destroys the checked-out Workerd client before joined-org Bento requests', async () => { + const lifecycleTrace: string[] = [] + let stateRead = 0 + pgQueryMock.mockImplementation(async (query) => { + if (String(query).includes('FROM public.role_bindings AS binding')) { + lifecycleTrace.push('binding:queried') + return { rows: [activeBinding()] } + } + lifecycleTrace.push(stateRead++ === 0 ? 'state:preflight' : 'state:reconciled') + return { rows: [firstOrgDatabaseState()] } + }) + // Workerd closeClient is intentionally a no-op. The checked-out client + // must be destroyed explicitly before any external Bento request starts. + closeClientMock.mockImplementation(async () => undefined) + pgReleaseMock.mockImplementation((destroy) => { + lifecycleTrace.push(`client:released:${destroy}`) + }) + syncBentoSubscriberTagsMock.mockImplementation(async () => { + lifecycleTrace.push('tag:suppress') + return true + }) + trackBentoEventMock.mockImplementation(async () => { + lifecycleTrace.push('event:joined') + return true + }) + + await syncRoleBinding() + + expect(lifecycleTrace).toEqual([ + 'binding:queried', + 'client:released:true', + 'state:preflight', + 'client:released:true', + 'tag:suppress', + 'event:joined', + 'state:reconciled', + 'client:released:true', + ]) + expect(getPgClientMock).toHaveBeenCalledOnce() + expect(pgConnectMock).toHaveBeenCalledTimes(3) + expect(pgReleaseMock).toHaveBeenCalledTimes(3) + expect(pgReleaseMock).toHaveBeenNthCalledWith(1, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(2, true) + expect(pgReleaseMock).toHaveBeenNthCalledWith(3, true) + expect(closeClientMock).toHaveBeenCalledOnce() + expect(closeClientMock).toHaveBeenCalledWith(expect.anything(), getPgClientMock.mock.results[0]?.value) + expect(trackBentoEventMock.mock.invocationCallOrder[0]) + .toBeLessThan(closeClientMock.mock.invocationCallOrder[0]!) + }) + + it.each([ + ['deleted binding', null], + ['expired binding', activeBinding({ is_active: false })], + ['API-key binding', activeBinding({ principal_type: 'apikey' })], + ['non-org binding', activeBinding({ scope_type: 'app' })], + ['null-org binding', activeBinding({ org_id: null })], + ['inherited binding', activeBinding({ is_direct: false })], + ['binding without a current user email', activeBinding({ email: null })], + ])('is a no-op for a %s', async (_label, binding) => { + pgQueryMock.mockResolvedValue({ rows: binding ? [binding] : [] }) + + await syncRoleBinding() + + expect(syncBentoSubscriberTagsMock).not.toHaveBeenCalled() + expect(trackBentoEventMock).not.toHaveBeenCalled() + }) + + it('repeats the fact event while keeping permanent suppression idempotent for duplicate messages', async () => { + await syncRoleBinding() + await syncRoleBinding() + + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledTimes(2) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(1, expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'current.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(syncBentoSubscriberTagsMock).toHaveBeenNthCalledWith(2, expect.anything(), { + deleteSegments: [LIFECYCLE_TAG], + email: 'current.user@example.com', + segments: [SUPPRESSION_TAG], + }) + expect(trackBentoEventMock).toHaveBeenCalledTimes(2) + const factData = { + joined_at: JOINED_AT, + org_id: ORG_ID, + role_binding_id: ROLE_BINDING_ID, + user_id: USER_ID, + } + expect(trackBentoEventMock).toHaveBeenNthCalledWith( + 1, + expect.anything(), + 'current.user@example.com', + factData, + 'user:joined_org', + ) + expect(trackBentoEventMock).toHaveBeenNthCalledWith( + 2, + expect.anything(), + 'current.user@example.com', + factData, + 'user:joined_org', + ) + }) + + it.each([ + ['tag delivery returns false', () => syncBentoSubscriberTagsMock.mockResolvedValueOnce(false)], + ['tag delivery throws', () => syncBentoSubscriberTagsMock.mockRejectedValueOnce(new Error('Bento unavailable'))], + ['fact event returns false', () => trackBentoEventMock.mockResolvedValueOnce(false)], + ])('fails for retry when configured Bento %s', async (_label, configureFailure) => { + configureFailure() + + await expect(syncRoleBinding()).rejects.toThrow() + }) + + it.each([ + ['tag delivery returns false', () => syncBentoSubscriberTagsMock.mockResolvedValueOnce(false), false], + ['tag delivery throws', () => syncBentoSubscriberTagsMock.mockRejectedValueOnce(new Error('Bento unavailable')), false], + ['fact event returns false', () => trackBentoEventMock.mockResolvedValueOnce(false), true], + ['fact event throws', () => trackBentoEventMock.mockRejectedValueOnce(new Error('Bento unavailable')), true], + ])('reconciles deletion after ambiguous joined-org %s', async (_label, configureFailure, eventAttempted) => { + pgQueryMock + .mockResolvedValueOnce({ rows: [activeBinding()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(true)] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + configureFailure() + + await expect(syncRoleBinding()).rejects.toThrow() + + expect(trackBentoEventMock).toHaveBeenCalledTimes(eventAttempted ? 1 : 0) + expect(syncBentoSubscriberTagsMock).toHaveBeenLastCalledWith(expect.anything(), [{ + deleteSegments: [LIFECYCLE_TAG], + email: 'current.user@example.com', + segments: [SUPPRESSION_TAG], + }], expect.any(AbortSignal)) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'current.user@example.com', expect.any(AbortSignal)) + }) + + it('preserves both errors when a Bento mutation and deletion reconciliation fail', async () => { + const mutationError = new Error('Bento mutation unavailable') + const reconciliationError = new Error('Bento reconciliation unavailable') + pgQueryMock + .mockResolvedValueOnce({ rows: [activeBinding()] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(true)] }) + .mockResolvedValueOnce({ rows: [firstOrgDatabaseState(false, false)] }) + syncBentoSubscriberTagsMock + .mockRejectedValueOnce(mutationError) + .mockRejectedValueOnce(reconciliationError) + + const result = syncRoleBinding() + + await expect(result).rejects.toMatchObject({ + errors: [mutationError, reconciliationError], + message: 'Bento mutation and first-organization reconciliation failed', + }) + expect(unsubscribeBentoMock).toHaveBeenCalledWith( + expect.anything(), + 'current.user@example.com', + expect.any(AbortSignal), + ) + }) + + it('succeeds as a no-op when Bento is not configured', async () => { + syncBentoSubscriberTagsMock.mockResolvedValue(undefined) + trackBentoEventMock.mockResolvedValue(undefined) + + await expect(syncRoleBinding()).resolves.toBeUndefined() + }) +}) diff --git a/tests/bento-first-org-route.unit.test.ts b/tests/bento-first-org-route.unit.test.ts new file mode 100644 index 0000000000..4f7a51cd9d --- /dev/null +++ b/tests/bento-first-org-route.unit.test.ts @@ -0,0 +1,151 @@ +import { readFile } from 'node:fs/promises' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { + syncBentoFirstOrgOnRoleBindingWriteMock, + syncBentoFirstOrgOnUserCreateMock, +} = vi.hoisted(() => ({ + syncBentoFirstOrgOnRoleBindingWriteMock: vi.fn(async () => undefined), + syncBentoFirstOrgOnUserCreateMock: vi.fn(async () => undefined), +})) + +vi.mock('../supabase/functions/_backend/utils/bento_first_org.ts', () => ({ + syncBentoFirstOrgOnRoleBindingWrite: syncBentoFirstOrgOnRoleBindingWriteMock, + syncBentoFirstOrgOnUserCreate: syncBentoFirstOrgOnUserCreateMock, +})) + +const apiWorker = (await import('../cloudflare_workers/api/index.ts')).default + +const API_SECRET = 'test-secret' +const ORG_ID = '22222222-2222-4222-8222-222222222222' +const ROLE_BINDING_ID = '33333333-3333-4333-8333-333333333333' +const ROLE_ID = '44444444-4444-4444-8444-444444444444' +const USER_ID = '11111111-1111-4111-8111-111111111111' +const originalApiSecret = process.env.API_SECRET + +const roleBindingRecord = { + app_id: null, + bundle_id: null, + channel_id: null, + expires_at: null, + granted_at: '2026-08-03T09:15:00.000Z', + granted_by: USER_ID, + id: ROLE_BINDING_ID, + is_direct: true, + org_id: ORG_ID, + parent_binding_id: null, + principal_id: USER_ID, + principal_type: 'user', + reason: 'Accepted invitation', + role_id: ROLE_ID, + scope_type: 'org', +} + +function requestPayload(payload: unknown, apiSecret: string | null = API_SECRET) { + const headers: Record = { + 'content-type': 'application/json', + 'x-capgo-queue-max-reads': '5', + 'x-capgo-queue-name': 'on_user_org_access', + 'x-capgo-queue-read-count': '1', + } + if (apiSecret !== null) + headers.apisecret = apiSecret + + return apiWorker.fetch(new Request('https://api.capgo.app/triggers/on_user_org_access', { + body: JSON.stringify(payload), + headers, + method: 'POST', + })) +} + +function requestRoleBindingWrite(type: 'INSERT' | 'UPDATE', apiSecret: string | null = API_SECRET) { + return requestPayload({ + old_record: type === 'UPDATE' ? roleBindingRecord : null, + record: roleBindingRecord, + schema: 'public', + table: 'role_bindings', + type, + }, apiSecret) +} + +describe('first-organization lifecycle trigger route', () => { + beforeEach(() => { + process.env.API_SECRET = API_SECRET + vi.clearAllMocks() + syncBentoFirstOrgOnRoleBindingWriteMock.mockResolvedValue(undefined) + }) + + afterEach(() => { + if (originalApiSecret === undefined) + delete process.env.API_SECRET + else + process.env.API_SECRET = originalApiSecret + }) + + it.each(['INSERT', 'UPDATE'] as const)('dispatches authenticated %s payloads and returns JSON success', async (type) => { + const response = await requestRoleBindingWrite(type) + + expect(response.status).toBe(200) + expect(response.headers.get('content-type')).toContain('application/json') + await expect(response.json()).resolves.toEqual({ status: 'ok' }) + expect(syncBentoFirstOrgOnRoleBindingWriteMock).toHaveBeenCalledOnce() + expect(syncBentoFirstOrgOnRoleBindingWriteMock).toHaveBeenCalledWith(expect.anything(), ROLE_BINDING_ID) + }) + + it('returns 5xx so the queue retries when lifecycle delivery fails', async () => { + syncBentoFirstOrgOnRoleBindingWriteMock.mockRejectedValueOnce(new Error('Bento unavailable')) + + const response = await requestRoleBindingWrite('INSERT') + + expect(response.status).toBeGreaterThanOrEqual(500) + expect(response.status).toBeLessThan(600) + expect(syncBentoFirstOrgOnRoleBindingWriteMock).toHaveBeenCalledOnce() + }) + + it.each([ + ['a missing API secret', null], + ['an invalid API secret', 'wrong-secret'], + ])('rejects %s before invoking the lifecycle helper', async (_label, apiSecret) => { + const response = await requestRoleBindingWrite('INSERT', apiSecret) + + expect(response.status).toBe(400) + expect(syncBentoFirstOrgOnRoleBindingWriteMock).not.toHaveBeenCalled() + }) + + it.each([ + ['null', null], + ['an array', []], + ['a string', 'role_bindings'], + ['a number', 42], + ['a boolean', false], + ])('rejects %s JSON payload with a controlled validation error', async (_label, payload) => { + const response = await requestPayload(payload) + + expect(response.status).toBe(400) + await expect(response.json()).resolves.toMatchObject({ + error: 'invalid_payload', + }) + expect(syncBentoFirstOrgOnRoleBindingWriteMock).not.toHaveBeenCalled() + }) + + it.each([ + ['a different table', 'table_not_match', { old_record: null, record: roleBindingRecord, schema: 'public', table: 'users', type: 'INSERT' }], + ['a delete event', 'type_not_match', { old_record: null, record: roleBindingRecord, schema: 'public', table: 'role_bindings', type: 'DELETE' }], + ['a missing record ID', 'invalid_payload', { old_record: null, record: {}, schema: 'public', table: 'role_bindings', type: 'INSERT' }], + ['a malformed record ID', 'invalid_payload', { old_record: null, record: { ...roleBindingRecord, id: 'not-a-uuid' }, schema: 'public', table: 'role_bindings', type: 'UPDATE' }], + ])('rejects %s with a controlled validation error', async (_label, error, payload) => { + const response = await requestPayload(payload) + + expect(response.status).toBe(400) + await expect(response.json()).resolves.toMatchObject({ error }) + expect(syncBentoFirstOrgOnRoleBindingWriteMock).not.toHaveBeenCalled() + }) + + it('registers the same route in the Supabase trigger router', async () => { + // Importing this deployment entry point evaluates Deno.serve and starts a + // server, so verify its static wiring without executing the module. + const source = await readFile(new URL('../supabase/functions/triggers/index.ts', import.meta.url), 'utf8') + + expect(source).toContain('appGlobal.route(\'/on_user_org_access\', on_user_org_access)') + }) +}) diff --git a/tests/bento-response-acceptance.unit.test.ts b/tests/bento-response-acceptance.unit.test.ts new file mode 100644 index 0000000000..55a9978ef1 --- /dev/null +++ b/tests/bento-response-acceptance.unit.test.ts @@ -0,0 +1,139 @@ +import type { Context } from 'hono' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { syncBentoSubscriberTags, trackBentoEvent, unsubscribeBento } from '../supabase/functions/_backend/utils/bento.ts' + +vi.mock('../supabase/functions/_backend/utils/logging.ts', () => ({ + cloudlog: vi.fn(), + cloudlogErr: vi.fn(), + serializeError: (error: unknown) => error, +})) + +vi.mock('../supabase/functions/_backend/utils/utils.ts', () => ({ + getEnv: (_context: unknown, key: string) => { + const values: Record = { + BENTO_PUBLISHABLE_KEY: 'publishable-key-value', + BENTO_SECRET_KEY: 'secret-key-value', + BENTO_SITE_UUID: 'site-uuid-value', + } + return values[key] ?? '' + }, +})) + +const fetchMock = vi.fn<( + input: string | URL | Request, + init?: RequestInit, +) => Promise>() + +function createContext() { + return { + get: vi.fn(() => 'request-id'), + } as unknown as Context +} + +function jsonResponse(body: unknown) { + return new Response(JSON.stringify(body), { + headers: { 'content-type': 'application/json' }, + status: 200, + }) +} + +function queueAcknowledgement(body: unknown) { + fetchMock.mockResolvedValueOnce(jsonResponse(body)) +} + +const subscriberUpdates = [ + { + deleteSegments: [], + email: 'first.user@example.com', + segments: ['onboarding:awaiting_first_org'], + }, + { + deleteSegments: ['onboarding:awaiting_first_org'], + email: 'second.user@example.com', + segments: ['onboarding:first_org_recovery_suppressed'], + }, +] + +describe('configured Bento response acceptance', () => { + beforeEach(() => { + fetchMock.mockReset() + vi.stubGlobal('fetch', fetchMock) + }) + + afterEach(() => { + vi.unstubAllGlobals() + }) + + describe('syncBentoSubscriberTags', () => { + it('accepts an acknowledgement for the exact subscriber count', async () => { + queueAcknowledgement({ failed: 0, results: 2 }) + + await expect(syncBentoSubscriberTags(createContext(), subscriberUpdates)).resolves.toBe(true) + }) + + it.each([ + ['zero results', { failed: 0, results: 0 }], + ['a missing result count', { failed: 0 }], + ['a missing failure count', { results: 2 }], + ['a short result count', { failed: 0, results: 1 }], + ['a malformed result count', { failed: 0, results: '2' }], + ['a malformed acknowledgement', null], + ['a non-zero failure count', { failed: 1, results: 2 }], + ])('rejects %s', async (_label, acknowledgement) => { + queueAcknowledgement(acknowledgement) + + await expect(syncBentoSubscriberTags(createContext(), subscriberUpdates)).resolves.toBe(false) + }) + }) + + describe('trackBentoEvent', () => { + it('accepts an acknowledgement for exactly one event', async () => { + queueAcknowledgement({ failed: 0, results: 1 }) + + await expect(trackBentoEvent( + createContext(), + 'event.user@example.com', + { source: 'unit-test' }, + 'user:created', + )).resolves.toBe(true) + }) + + it.each([ + ['zero results', { failed: 0, results: 0 }], + ['a missing result count', { failed: 0 }], + ['a missing failure count', { results: 1 }], + ['a malformed result count', { failed: 0, results: '1' }], + ['a malformed acknowledgement', null], + ['a non-zero failure count', { failed: 1, results: 1 }], + ])('rejects %s', async (_label, acknowledgement) => { + queueAcknowledgement(acknowledgement) + + await expect(trackBentoEvent( + createContext(), + 'event.user@example.com', + { source: 'unit-test' }, + 'user:created', + )).resolves.toBe(false) + }) + }) + + describe('unsubscribeBento', () => { + it('accepts an acknowledgement for exactly one command', async () => { + queueAcknowledgement({ results: 1 }) + + await expect(unsubscribeBento(createContext(), 'unsubscribed.user@example.com')).resolves.toBe(true) + }) + + it.each([ + ['zero results', { results: 0 }], + ['a missing result count', {}], + ['a malformed result count', { results: '1' }], + ['a malformed acknowledgement', null], + ])('rejects %s', async (_label, acknowledgement) => { + queueAcknowledgement(acknowledgement) + + await expect(unsubscribeBento(createContext(), 'unsubscribed.user@example.com')).resolves.toBe(false) + }) + }) +}) diff --git a/tests/create-api-key-deletion-race.unit.test.ts b/tests/create-api-key-deletion-race.unit.test.ts new file mode 100644 index 0000000000..9303ae53e2 --- /dev/null +++ b/tests/create-api-key-deletion-race.unit.test.ts @@ -0,0 +1,125 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + closeClientMock, + getPgClientMock, + pgConnectMock, + pgQueryMock, + pgReleaseMock, +} = vi.hoisted(() => { + const pgQueryMock = vi.fn() + const pgReleaseMock = vi.fn() + const pgConnectMock = vi.fn(async () => ({ query: pgQueryMock, release: pgReleaseMock })) + return { + closeClientMock: vi.fn(), + getPgClientMock: vi.fn(() => ({ connect: pgConnectMock })), + pgConnectMock, + pgQueryMock, + pgReleaseMock, + } +}) + +vi.mock('../supabase/functions/_backend/utils/pg.ts', () => ({ + closeClient: closeClientMock, + getPgClient: getPgClientMock, +})) + +const { createApiKey } = await import('../supabase/functions/_backend/utils/supabase.ts') + +const USER_ID = '11111111-1111-4111-8111-111111111111' + +function normalizedQuery(query: unknown) { + return String(query).replace(/\s+/g, ' ').trim() +} + +function context() { + return { get: vi.fn(() => 'request-id') } as never +} + +describe('createApiKey account-deletion serialization', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('locks the public profile and rolls back before API-key reads when deletion is scheduled', async () => { + pgQueryMock.mockImplementation(async (query: unknown) => { + const sql = normalizedQuery(query) + if (sql === 'BEGIN' || sql === 'ROLLBACK' || sql.startsWith('SET LOCAL lock_timeout')) + return { rowCount: null, rows: [] } + if (sql.includes('SELECT id FROM public.users')) + return { rowCount: 1, rows: [{ id: USER_ID }] } + if (sql.includes('FROM public.to_delete_accounts')) + return { rowCount: 1, rows: [{ deletion_scheduled: true }] } + throw new Error(`Unexpected query: ${sql}`) + }) + + await createApiKey(context(), USER_ID) + + const queries = pgQueryMock.mock.calls.map(([query]) => normalizedQuery(query)) + expect(queries).toEqual([ + 'BEGIN', + `SET LOCAL lock_timeout = '5s'`, + 'SELECT id FROM public.users WHERE id = $1::uuid FOR UPDATE', + 'SELECT EXISTS ( SELECT 1 FROM public.to_delete_accounts WHERE account_id = $1::uuid ) AS deletion_scheduled', + 'ROLLBACK', + ]) + expect(queries.some(query => query.includes('FROM public.apikeys'))).toBe(false) + expect(pgConnectMock).toHaveBeenCalledOnce() + expect(pgReleaseMock).toHaveBeenCalledWith(true) + expect(closeClientMock).toHaveBeenCalledOnce() + }) + + it('continues default-key provisioning only after the deletion guard is clear', async () => { + pgQueryMock.mockImplementation(async (query: unknown) => { + const sql = normalizedQuery(query) + if (sql === 'BEGIN' || sql === 'COMMIT' || sql.startsWith('SET LOCAL lock_timeout')) + return { rowCount: null, rows: [] } + if (sql.includes('SELECT id FROM public.users')) + return { rowCount: 1, rows: [{ id: USER_ID }] } + if (sql.includes('FROM public.to_delete_accounts')) + return { rowCount: 1, rows: [{ deletion_scheduled: false }] } + if (sql.includes('SELECT count(*)::text AS count FROM public.apikeys')) + return { rowCount: 1, rows: [{ count: '0' }] } + if (sql.startsWith('SELECT DISTINCT rb.org_id')) + return { rowCount: 1, rows: [{ org_id: '22222222-2222-4222-8222-222222222222' }] } + if (sql.startsWith('WITH default_keys')) + return { rowCount: 3, rows: [] } + throw new Error(`Unexpected query: ${sql}`) + }) + + await createApiKey(context(), USER_ID) + + const queries = pgQueryMock.mock.calls.map(([query]) => normalizedQuery(query)) + const guardIndex = queries.findIndex(query => query.includes('FROM public.to_delete_accounts')) + const insertIndex = queries.findIndex(query => query.startsWith('WITH default_keys')) + expect(guardIndex).toBeGreaterThan(0) + expect(insertIndex).toBeGreaterThan(guardIndex) + expect(queries.at(-1)).toBe('COMMIT') + expect(pgConnectMock).toHaveBeenCalledOnce() + expect(pgReleaseMock).toHaveBeenCalledWith(true) + expect(closeClientMock).toHaveBeenCalledOnce() + }) + + it('rolls back, destroys the client, and propagates a lock timeout for queue retry', async () => { + const lockTimeout = Object.assign(new Error('canceling statement due to lock timeout'), { code: '55P03' }) + pgQueryMock.mockImplementation(async (query: unknown) => { + const sql = normalizedQuery(query) + if (sql === 'BEGIN' || sql === 'ROLLBACK' || sql.startsWith('SET LOCAL lock_timeout')) + return { rowCount: null, rows: [] } + if (sql.includes('SELECT id FROM public.users')) + throw lockTimeout + throw new Error(`Unexpected query: ${sql}`) + }) + + await expect(createApiKey(context(), USER_ID)).rejects.toBe(lockTimeout) + + expect(pgQueryMock.mock.calls.map(([query]) => normalizedQuery(query))).toEqual([ + 'BEGIN', + `SET LOCAL lock_timeout = '5s'`, + 'SELECT id FROM public.users WHERE id = $1::uuid FOR UPDATE', + 'ROLLBACK', + ]) + expect(pgReleaseMock).toHaveBeenCalledWith(true) + expect(closeClientMock).toHaveBeenCalledOnce() + }) +}) diff --git a/tests/on-user-delete-bento.unit.test.ts b/tests/on-user-delete-bento.unit.test.ts new file mode 100644 index 0000000000..d375ca4668 --- /dev/null +++ b/tests/on-user-delete-bento.unit.test.ts @@ -0,0 +1,297 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + cancelSubscriptionMock, + supabaseAdminMock, + syncBentoSubscriberTagsMock, + unsubscribeBentoMock, +} = vi.hoisted(() => ({ + cancelSubscriptionMock: vi.fn(async () => undefined as boolean | undefined), + supabaseAdminMock: vi.fn(), + syncBentoSubscriberTagsMock: vi.fn<( + c: unknown, + update: unknown, + signal?: AbortSignal, + ) => Promise>(async () => true), + unsubscribeBentoMock: vi.fn<( + c: unknown, + email: string, + signal?: AbortSignal, + ) => Promise>(async () => true), +})) + +vi.mock('../supabase/functions/_backend/utils/bento.ts', () => ({ + syncBentoSubscriberTags: syncBentoSubscriberTagsMock, + trackBentoEvent: vi.fn(), + unsubscribeBento: unsubscribeBentoMock, +})) + +vi.mock('../supabase/functions/_backend/utils/hono.ts', async () => { + const actual = await vi.importActual('../supabase/functions/_backend/utils/hono.ts') + return { + ...actual, + middlewareAPISecret: async (_c: unknown, next: () => Promise) => await next(), + } +}) + +vi.mock('../supabase/functions/_backend/utils/stripe.ts', () => ({ + cancelSubscription: cancelSubscriptionMock, +})) + +vi.mock('../supabase/functions/_backend/utils/supabase.ts', () => ({ + supabaseAdmin: supabaseAdminMock, +})) + +const { app } = await import('../supabase/functions/_backend/triggers/on_user_delete.ts') +const { BENTO_DELETED_USER_OPERATION_TIMEOUT_MS } = await import('../supabase/functions/_backend/utils/bento_first_org.ts') + +const USER_ID = '11111111-1111-4111-8111-111111111111' + +function queryBuilder(data: unknown[] = [], error: unknown = null) { + const result = Promise.resolve({ data, error }) + const builder = { + eq: vi.fn(), + in: vi.fn(), + select: vi.fn(), + then: result.then.bind(result), + } + builder.eq.mockReturnValue(builder) + builder.in.mockReturnValue(builder) + builder.select.mockReturnValue(builder) + return builder +} + +function configureSingleOrgCleanup() { + const orgId = '22222222-2222-4222-8222-222222222222' + const roleBindingResults = [ + [{ expires_at: null, org_id: orgId }], + [{ expires_at: null, org_id: orgId, principal_id: USER_ID }], + [], + ] + let roleBindingRead = 0 + const client = { + from: vi.fn((table: string) => { + if (table === 'role_bindings') + return queryBuilder(roleBindingResults[roleBindingRead++] ?? []) + if (table === 'orgs') + return queryBuilder([{ customer_id: 'cus_deleted_user', id: orgId, management_email: null }]) + return queryBuilder() + }), + storage: { + from: vi.fn(() => ({ + list: vi.fn(async () => ({ data: [] })), + })), + }, + } + supabaseAdminMock.mockReturnValue(client) +} + +function deletedUserRecord() { + return { + ban_time: null, + country: null, + created_at: '2026-08-03T08:30:00.000Z', + created_via_invite: false, + discord_username: null, + email: ' Deleted.User@Example.COM ', + email_preferences: {}, + enable_notifications: false, + first_name: 'Deleted', + format_locale: null, + github_id: null, + github_username: null, + id: USER_ID, + image_url: null, + last_name: 'User', + opt_for_newsletters: false, + updated_at: '2026-08-03T09:00:00.000Z', + } +} + +function postDelete() { + return app.request('http://local/', { + body: JSON.stringify({ + old_record: deletedUserRecord(), + record: null, + schema: 'public', + table: 'users', + type: 'DELETE', + }), + headers: { 'content-type': 'application/json' }, + method: 'POST', + }) +} + +describe('user deletion Bento recovery safety', () => { + beforeEach(() => { + vi.clearAllMocks() + cancelSubscriptionMock.mockResolvedValue(undefined) + syncBentoSubscriberTagsMock.mockResolvedValue(true) + unsubscribeBentoMock.mockResolvedValue(true) + const emptyBuilder = queryBuilder() + supabaseAdminMock.mockReturnValue({ + from: vi.fn(() => emptyBuilder), + }) + }) + + it('suppresses and unsubscribes a no-org user before organization cleanup returns', async () => { + const response = await postDelete() + + expect(response.status).toBe(200) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledWith(expect.anything(), [{ + deleteSegments: ['onboarding:awaiting_first_org'], + email: 'deleted.user@example.com', + segments: ['onboarding:first_org_recovery_suppressed'], + }], expect.any(AbortSignal)) + expect(unsubscribeBentoMock).toHaveBeenCalledWith(expect.anything(), 'deleted.user@example.com', expect.any(AbortSignal)) + expect(syncBentoSubscriberTagsMock.mock.invocationCallOrder[0]) + .toBeLessThan(unsubscribeBentoMock.mock.invocationCallOrder[0]) + expect(cancelSubscriptionMock).not.toHaveBeenCalled() + }) + + it.each([ + ['suppression', () => syncBentoSubscriberTagsMock.mockResolvedValue(false)], + ['suppression exception', () => syncBentoSubscriberTagsMock.mockRejectedValue(new Error('Bento unavailable'))], + ['unsubscribe', () => unsubscribeBentoMock.mockResolvedValue(false)], + ['unsubscribe exception', () => unsubscribeBentoMock.mockRejectedValue(new Error('Bento unavailable'))], + ])('fails for queue retry when Bento %s cleanup fails', async (_label, fail) => { + fail() + + const response = await postDelete() + + expect(response.status).toBe(500) + expect(unsubscribeBentoMock).toHaveBeenCalledOnce() + expect(syncBentoSubscriberTagsMock.mock.invocationCallOrder[0]) + .toBeLessThan(unsubscribeBentoMock.mock.invocationCallOrder[0]) + expect(supabaseAdminMock).toHaveBeenCalled() + }) + + it('completes subscription cleanup before returning a retryable Bento failure', async () => { + syncBentoSubscriberTagsMock.mockResolvedValue(false) + configureSingleOrgCleanup() + + const response = await postDelete() + + expect(response.status).toBe(500) + expect(cancelSubscriptionMock).toHaveBeenCalledWith(expect.anything(), 'cus_deleted_user') + }) + + it('returns a retryable failure when an RBAC lookup fails', async () => { + const failedBuilder = queryBuilder([], new Error('Database unavailable')) + supabaseAdminMock.mockReturnValue({ + from: vi.fn(() => failedBuilder), + }) + + const response = await postDelete() + + expect(response.status).toBe(500) + expect(cancelSubscriptionMock).not.toHaveBeenCalled() + }) + + it('returns a retryable failure when Stripe cleanup reports a failure', async () => { + cancelSubscriptionMock.mockResolvedValue(false) + configureSingleOrgCleanup() + + const response = await postDelete() + + expect(response.status).toBe(500) + expect(cancelSubscriptionMock).toHaveBeenCalledWith(expect.anything(), 'cus_deleted_user') + }) + + it('runs subscription cleanup while Bento suppression is still pending', async () => { + let resolveSuppression!: (value: boolean) => void + syncBentoSubscriberTagsMock.mockImplementationOnce(async () => await new Promise((resolve) => { + resolveSuppression = resolve + })) + configureSingleOrgCleanup() + + const responsePromise = postDelete() + + await vi.waitFor(() => { + expect(cancelSubscriptionMock).toHaveBeenCalledWith(expect.anything(), 'cus_deleted_user') + }) + expect(unsubscribeBentoMock).not.toHaveBeenCalled() + + resolveSuppression(true) + const response = await responsePromise + expect(response.status).toBe(200) + expect(unsubscribeBentoMock).toHaveBeenCalledOnce() + }) + + it('still submits unsubscribe and returns a retryable failure when suppression hangs', async () => { + vi.useFakeTimers() + try { + const lifecycleTrace: string[] = [] + let suppressionSignal: AbortSignal | undefined + let suppressionSettled = false + syncBentoSubscriberTagsMock.mockImplementationOnce(async (_c, _update, signal) => { + suppressionSignal = signal + lifecycleTrace.push('suppression:started') + await new Promise((resolve) => { + signal?.addEventListener('abort', () => { + lifecycleTrace.push('suppression:aborted') + queueMicrotask(resolve) + }, { once: true }) + }) + suppressionSettled = true + lifecycleTrace.push('suppression:settled') + return false + }) + unsubscribeBentoMock.mockImplementationOnce(async () => { + expect(suppressionSignal?.aborted).toBe(true) + expect(suppressionSettled).toBe(true) + lifecycleTrace.push('unsubscribe:started') + return true + }) + + const responsePromise = postDelete() + await vi.advanceTimersByTimeAsync(BENTO_DELETED_USER_OPERATION_TIMEOUT_MS + 1) + + const response = await responsePromise + expect(response.status).toBe(500) + expect(unsubscribeBentoMock).toHaveBeenCalledOnce() + expect(supabaseAdminMock).toHaveBeenCalled() + expect(lifecycleTrace).toEqual([ + 'suppression:started', + 'suppression:aborted', + 'suppression:settled', + 'unsubscribe:started', + ]) + } + finally { + vi.useRealTimers() + } + }) + + it('bounds a hanging unsubscribe and returns a retryable failure', async () => { + vi.useFakeTimers() + try { + let unsubscribeSignal: AbortSignal | undefined + unsubscribeBentoMock.mockImplementationOnce(async (_c, _email, signal) => await new Promise((resolve) => { + unsubscribeSignal = signal + signal?.addEventListener('abort', () => resolve(false), { once: true }) + })) + + const responsePromise = postDelete() + await vi.advanceTimersByTimeAsync(BENTO_DELETED_USER_OPERATION_TIMEOUT_MS + 1) + + const response = await responsePromise + expect(response.status).toBe(500) + expect(syncBentoSubscriberTagsMock).toHaveBeenCalledOnce() + expect(unsubscribeBentoMock).toHaveBeenCalledOnce() + expect(unsubscribeSignal?.aborted).toBe(true) + } + finally { + vi.useRealTimers() + } + }) + + it('continues as a no-op when Bento is not configured', async () => { + syncBentoSubscriberTagsMock.mockResolvedValue(undefined) + unsubscribeBentoMock.mockResolvedValue(undefined) + + const response = await postDelete() + + expect(response.status).toBe(200) + }) +}) diff --git a/tests/on-user-update-bento.unit.test.ts b/tests/on-user-update-bento.unit.test.ts new file mode 100644 index 0000000000..97b097fe03 --- /dev/null +++ b/tests/on-user-update-bento.unit.test.ts @@ -0,0 +1,258 @@ +import { readFile } from 'node:fs/promises' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + cleanStoredImageMetadataMock, + createApiKeyMock, + syncBentoFirstOrgOnEmailChangeMock, + syncUserPreferenceTagsMock, +} = vi.hoisted(() => ({ + cleanStoredImageMetadataMock: vi.fn(async () => undefined), + createApiKeyMock: vi.fn(async () => undefined), + syncBentoFirstOrgOnEmailChangeMock: vi.fn(async () => true as boolean | undefined), + syncUserPreferenceTagsMock: vi.fn(async () => undefined), +})) + +vi.mock('../supabase/functions/_backend/utils/bento_first_org.ts', () => ({ + normalizeBentoEmail: (email: string) => email.trim().toLowerCase(), + syncBentoFirstOrgOnEmailChange: syncBentoFirstOrgOnEmailChangeMock, +})) + +vi.mock('../supabase/functions/_backend/utils/hono.ts', async () => { + const actual = await vi.importActual('../supabase/functions/_backend/utils/hono.ts') + return { + ...actual, + middlewareAPISecret: async (_c: unknown, next: () => Promise) => await next(), + } +}) + +vi.mock('../supabase/functions/_backend/utils/image.ts', () => ({ + cleanStoredImageMetadata: cleanStoredImageMetadataMock, +})) + +vi.mock('../supabase/functions/_backend/utils/supabase.ts', () => ({ + createApiKey: createApiKeyMock, +})) + +vi.mock('../supabase/functions/_backend/utils/user_preferences.ts', () => ({ + syncUserPreferenceTags: syncUserPreferenceTagsMock, +})) + +const { app } = await import('../supabase/functions/_backend/triggers/on_user_update.ts') + +const USER_ID = '11111111-1111-4111-8111-111111111111' + +function userRecord(email: string, overrides: Record = {}) { + return { + ban_time: null, + country: null, + created_at: '2026-08-03T08:30:00.000Z', + created_via_invite: false, + discord_username: null, + email, + email_preferences: {}, + enable_notifications: false, + first_name: 'User', + format_locale: null, + github_id: null, + github_username: null, + id: USER_ID, + image_url: null, + last_name: 'Example', + opt_for_newsletters: false, + updated_at: '2026-08-03T09:00:00.000Z', + ...overrides, + } +} + +async function postUpdate(record: ReturnType, oldRecord: ReturnType) { + return await app.request('http://local/', { + body: JSON.stringify({ + old_record: oldRecord, + record, + schema: 'public', + table: 'users', + type: 'UPDATE', + }), + headers: { 'content-type': 'application/json' }, + method: 'POST', + }) +} + +describe('user update Bento subscriber identity', () => { + beforeEach(() => { + vi.clearAllMocks() + syncBentoFirstOrgOnEmailChangeMock.mockResolvedValue(true) + }) + + it('suppresses both aliases before applying a simultaneous preference change', async () => { + const lifecycleTrace: string[] = [] + syncBentoFirstOrgOnEmailChangeMock.mockImplementation(async () => { + lifecycleTrace.push('aliases:suppressed') + }) + syncUserPreferenceTagsMock.mockImplementation(async () => { + lifecycleTrace.push('preferences:synced') + }) + const oldRecord = userRecord(' Old.User@Example.COM ', { enable_notifications: false }) + const record = userRecord(' New.User@Example.COM ', { enable_notifications: true }) + + const response = await postUpdate(record, oldRecord) + + expect(response.status).toBe(200) + expect(lifecycleTrace).toEqual(['aliases:suppressed', 'preferences:synced']) + expect(syncBentoFirstOrgOnEmailChangeMock).toHaveBeenCalledWith( + expect.anything(), + 'old.user@example.com', + 'new.user@example.com', + ) + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'new.user@example.com', + record, + oldRecord, + 'old.user@example.com', + ) + }) + + it('keeps the existing preference sync and sends no command when the email is unchanged', async () => { + const oldRecord = userRecord('same@example.com', { enable_notifications: false }) + const record = userRecord('same@example.com', { enable_notifications: true }) + + const response = await postUpdate(record, oldRecord) + + expect(response.status).toBe(200) + expect(syncBentoFirstOrgOnEmailChangeMock).not.toHaveBeenCalled() + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'same@example.com', + record, + oldRecord, + 'same@example.com', + ) + }) + + it('sends no command when old and new emails differ only by normalization', async () => { + const oldRecord = userRecord(' Same.User@Example.COM ') + const record = userRecord('same.user@example.com') + + const response = await postUpdate(record, oldRecord) + + expect(response.status).toBe(200) + expect(syncBentoFirstOrgOnEmailChangeMock).not.toHaveBeenCalled() + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'same.user@example.com', + record, + oldRecord, + 'same.user@example.com', + ) + }) + + it.each([ + ['returns false', () => syncBentoFirstOrgOnEmailChangeMock.mockResolvedValueOnce(false)], + ['throws', () => syncBentoFirstOrgOnEmailChangeMock.mockRejectedValueOnce(new Error('Bento unavailable'))], + ])('returns 5xx before preference sync when suppression delivery %s', async (_label, configureFailure) => { + configureFailure() + + const response = await postUpdate(userRecord('new@example.com'), userRecord('old@example.com')) + + expect(response.status).toBe(500) + expect(syncUserPreferenceTagsMock).not.toHaveBeenCalled() + }) + + it('syncs preferences with canonical aliases before image cleanup even when cleanup fails', async () => { + const lifecycleTrace: string[] = [] + syncBentoFirstOrgOnEmailChangeMock.mockImplementationOnce(async () => { + lifecycleTrace.push('aliases:suppressed') + return true + }) + syncUserPreferenceTagsMock.mockImplementationOnce(async () => { + lifecycleTrace.push('preferences:synced') + }) + createApiKeyMock.mockImplementationOnce(async () => { + lifecycleTrace.push('api-key:created') + }) + cleanStoredImageMetadataMock.mockImplementationOnce(async () => { + lifecycleTrace.push('image:cleanup') + throw new Error('Storage unavailable') + }) + const oldRecord = userRecord('old@example.com', { image_url: 'old/avatar.png' }) + const record = userRecord('new@example.com', { image_url: 'new/avatar.png' }) + + const response = await postUpdate(record, oldRecord) + + expect(response.status).toBe(500) + expect(lifecycleTrace).toEqual(['aliases:suppressed', 'preferences:synced', 'api-key:created', 'image:cleanup']) + expect(syncBentoFirstOrgOnEmailChangeMock).toHaveBeenCalledWith( + expect.anything(), + 'old@example.com', + 'new@example.com', + ) + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'new@example.com', + record, + oldRecord, + 'old@example.com', + ) + }) + + it('syncs preferences before API key creation even when API key creation fails', async () => { + const lifecycleTrace: string[] = [] + syncBentoFirstOrgOnEmailChangeMock.mockImplementationOnce(async () => { + lifecycleTrace.push('aliases:suppressed') + return true + }) + syncUserPreferenceTagsMock.mockImplementationOnce(async () => { + lifecycleTrace.push('preferences:synced') + }) + createApiKeyMock.mockImplementationOnce(async () => { + lifecycleTrace.push('api-key:create') + throw new Error('Database unavailable') + }) + + const response = await postUpdate(userRecord('new@example.com'), userRecord('old@example.com')) + + expect(response.status).toBe(500) + expect(lifecycleTrace).toEqual(['aliases:suppressed', 'preferences:synced', 'api-key:create']) + expect(cleanStoredImageMetadataMock).not.toHaveBeenCalled() + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'new@example.com', + expect.anything(), + expect.anything(), + 'old@example.com', + ) + }) + + it('continues preference sync when suppression is an unconfigured no-op', async () => { + syncBentoFirstOrgOnEmailChangeMock.mockResolvedValue(undefined) + const record = userRecord('new@example.com') + const oldRecord = userRecord('old@example.com') + + const response = await postUpdate(record, oldRecord) + + expect(response.status).toBe(200) + expect(syncUserPreferenceTagsMock).toHaveBeenCalledWith( + expect.anything(), + 'new@example.com', + record, + oldRecord, + 'old@example.com', + ) + }) + + it('contains no legacy Bento email migration command or helper', async () => { + const [bentoSource, triggerSource] = await Promise.all([ + readFile(new URL('../supabase/functions/_backend/utils/bento.ts', import.meta.url), 'utf8'), + readFile(new URL('../supabase/functions/_backend/triggers/on_user_update.ts', import.meta.url), 'utf8'), + ]) + const legacyCommand = ['change', 'email'].join('_') + const legacyHelper = ['change', 'Email', 'Bento'].join('') + + expect(bentoSource).not.toContain(legacyCommand) + expect(bentoSource).not.toContain(legacyHelper) + expect(triggerSource).not.toContain(legacyCommand) + expect(triggerSource).not.toContain(legacyHelper) + }) +}) diff --git a/tests/user-preference-bento-tags.unit.test.ts b/tests/user-preference-bento-tags.unit.test.ts index ee805de1d6..5440b269ea 100644 --- a/tests/user-preference-bento-tags.unit.test.ts +++ b/tests/user-preference-bento-tags.unit.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { syncUserPreferenceTags } from '../supabase/functions/_backend/utils/user_preferences.ts' const syncBentoSubscriberTagsMock = vi.hoisted(() => vi.fn(async () => true)) +const SUPPRESSION_TAG = 'onboarding:first_org_recovery_suppressed' vi.mock('../supabase/functions/_backend/utils/bento.ts', () => ({ syncBentoSubscriberTags: syncBentoSubscriberTagsMock, @@ -60,6 +61,22 @@ describe('syncUserPreferenceTags email type', () => { segments: ['email_type:professional'], deleteSegments: expect.arrayContaining(['email_type:personal', 'email_type:disposable']), })) + + interface SubscriberUpdate { + deleteSegments: string[] + segments: string[] + } + const calls = syncBentoSubscriberTagsMock.mock.calls as unknown as Array<[ + unknown, + SubscriberUpdate | SubscriberUpdate[], + ]> + for (const [, rawUpdate] of calls) { + const updates = Array.isArray(rawUpdate) ? rawUpdate : [rawUpdate] + for (const update of updates) { + expect(update.segments).not.toContain(SUPPRESSION_TAG) + expect(update.deleteSegments).not.toContain(SUPPRESSION_TAG) + } + } }) it('backfills the email type tag for a same-address preference update', async () => {