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,22 @@ Tag does NOT contain: credit_usage_disabled
 Tag does NOT contain: onboarding_disabled
 ```
 
+#### 3a. First-Organization Recovery
+
+**Entry event**: `user:registered_without_org`
+
+Capgo adds `onboarding:awaiting_first_org` when a direct registration still has no active organization access. Creating or joining an organization removes that tag. A normalized email-address change permanently adds `onboarding:first_org_recovery_suppressed` to both the old and new Bento aliases so asynchronous subscriber updates can never send this recovery email to a stale address.
+
+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.
+
 #### 4. Weekly Statistics
 
 **Events**: `user:weekly_stats`
@@ -159,6 +175,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 { 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'
@@ -15,6 +16,7 @@ app.post('/', middlewareAPISecret, triggerValidator('users', 'INSERT'), async (c
   await createApiKey(c, record.id)
   cloudlog({ requestId: c.get('requestId'), message: 'createCustomer stripe' })
   await syncUserPreferenceTags(c, 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_org_access.ts

@@ -0,0 +1,25 @@
+import type { MiddlewareKeyVariables } from '../utils/hono.ts'
+import { Hono } from 'hono/tiny'
+import { syncBentoFirstOrgOnRoleBindingWrite } from '../utils/bento_first_org.ts'
+import { BRES, middlewareAPISecret, parseBody, simpleError } from '../utils/hono.ts'
+
+interface RoleBindingWritePayload {
+  record?: { id?: string | null } | null
+  table?: string
+  type?: string
+}
+
+export const app = new Hono<MiddlewareKeyVariables>()
+
+app.post('/', middlewareAPISecret, async (c) => {
+  const payload = await parseBody<RoleBindingWritePayload>(c)
+  if (payload.table !== 'role_bindings')
+    throw simpleError('table_not_match', 'Not role_bindings', { payload })
+  if (payload.type !== 'INSERT' && payload.type !== 'UPDATE')
+    throw simpleError('type_not_match', 'Not INSERT or UPDATE', { payload })
+  if (!payload.record?.id)
+    throw simpleError('invalid_payload', 'Missing role binding id', { payload })
+
+  await syncBentoFirstOrgOnRoleBindingWrite(c, payload.record.id)
+  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 { 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,14 +22,24 @@ 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 = record.email.trim().toLowerCase()
+  const oldEmail = oldRecord?.email?.trim().toLowerCase()
+  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 createApiKey(c, record.id)
-  await syncUserPreferenceTags(c, record.email, record, oldRecord, oldRecord?.email)
 
   const newImagePath = record.image_url
   const oldImagePath = oldRecord?.image_url
   if (newImagePath && newImagePath !== oldImagePath) {
     await cleanStoredImageMetadata(c, newImagePath)
   }
 
+  await syncUserPreferenceTags(c, record.email, record, oldRecord, oldRecord?.email)
+
   return c.json(BRES)
 })

supabase/functions/_backend/utils/bento_first_org.ts

@@ -0,0 +1,171 @@
+import type { Context } from 'hono'
+import type { MiddlewareKeyVariables } from './hono.ts'
+import type { Database } from './supabase.types.ts'
+import { syncBentoSubscriberTags, trackBentoEvent } 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'
+
+interface CurrentRoleBinding {
+  email: string | null
+  granted_at: string
+  id: string
+  is_active: boolean
+  is_direct: boolean
+  org_id: string | null
+  principal_id: string
+  principal_type: string
+  scope_type: string
+}
+
+function normalizedEmail(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 setAwaitingFirstOrgTag(c: Context, email: string, awaiting: boolean) {
+  const result = await syncBentoSubscriberTags(c, {
+    email,
+    segments: awaiting ? [BENTO_AWAITING_FIRST_ORG_TAG] : [],
+    deleteSegments: awaiting ? [] : [BENTO_AWAITING_FIRST_ORG_TAG],
+  })
+  ensureBentoDelivery(result, awaiting ? 'add_awaiting_first_org_tag' : 'remove_awaiting_first_org_tag')
+}
+
+export async function syncBentoFirstOrgOnEmailChange(c: Context, oldEmail: string, newEmail: string) {
+  const aliases = [...new Set([oldEmail, newEmail].map(normalizedEmail).filter(Boolean))]
+  return await syncBentoSubscriberTags(c, aliases.map(email => ({
+    email,
+    segments: [BENTO_FIRST_ORG_RECOVERY_SUPPRESSED_TAG],
+    deleteSegments: [BENTO_AWAITING_FIRST_ORG_TAG],
+  })))
+}
+
+async function hasActiveDirectOrgAccess(pgPool: ReturnType<typeof getPgClient>, userId: string) {
+  const pgClient = await pgPool.connect()
+  try {
+    const result = await pgClient.query<{ id: string }>(
+      `SELECT id
+       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())
+       LIMIT 1`,
+      [userId],
+    )
+    return result.rows.length > 0
+  }
+  finally {
+    pgClient.release(true)
+  }
+}
+
+export async function syncBentoFirstOrgOnUserCreate(
+  c: Context<MiddlewareKeyVariables>,
+  user: Database['public']['Tables']['users']['Row'],
+) {
+  const email = normalizedEmail(user.email)
+
+  if (user.created_via_invite) {
+    await setAwaitingFirstOrgTag(c, email, false)
+    return
+  }
+
+  const pgPool = getPgClient(c)
+  try {
+    if (await hasActiveDirectOrgAccess(pgPool, user.id)) {
+      await setAwaitingFirstOrgTag(c, email, false)
+      return
+    }
+
+    await setAwaitingFirstOrgTag(c, email, true)
+
+    if (await hasActiveDirectOrgAccess(pgPool, user.id)) {
+      await setAwaitingFirstOrgTag(c, email, false)
+      return
+    }
+
+    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')
+  }
+  finally {
+    await closeClient(c, pgPool)
+  }
+}
+
+export async function syncBentoFirstOrgOnRoleBindingWrite(
+  c: Context<MiddlewareKeyVariables>,
+  roleBindingId: string,
+) {
+  const pgPool = getPgClient(c)
+  try {
+    let binding: CurrentRoleBinding | undefined
+    const pgClient = await pgPool.connect()
+    try {
+      const result = await pgClient.query<CurrentRoleBinding>(
+        `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 {
+      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 = normalizedEmail(binding.email)
+    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,
+    }, BENTO_JOINED_ORG_EVENT)
+    ensureBentoDelivery(result, 'joined_org_event')
+  }
+  finally {
+    await closeClient(c, pgPool)
+  }
+}

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)

supabase/migrations/20260803161022_on_user_org_access_queue.sql

@@ -0,0 +1,57 @@
+-- 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 new row that currently grants active direct
+--   organization access. 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'
+);
+
+UPDATE public.cron_tasks
+SET
+    target = (target::jsonb || '["on_user_org_access"]'::jsonb)::text,
+    updated_at = pg_catalog.now()
+WHERE
+    name = 'high_frequency_queues'
+    AND task_type = 'function_queue'::public.cron_task_type
+    AND NOT (target::jsonb ? 'on_user_org_access');