feat(backend): track first organization onboarding in Bento

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)

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)

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<MiddlewareKeyVariables>()
 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, {

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<MiddlewareKeyVariables>()
 
+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<unknown>[] = []
   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,74 +324,75 @@ 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)
   const now = new Date()
 
   // 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
     .map(binding => getBindingPrincipalId(binding))
     .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) => {

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<Database['public']['Tables']['role_bindings']['Row']>
+  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)
+})

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