feat(webhook,engines): account-scoped routing

PR 8 of the multi-user accounts series. Flips the WhatsApp webhook,
automations engine, and flows engine from per-user tenancy to
per-account so a shared inbox actually behaves like one.

Why this matters
- Pre-this-PR: an inbound message resolved the recipient by
  whatsapp_config.user_id and dispatched only that user's
  automations / flows. Teammates joining via an invite would see
  the message in the shared inbox but their automations / flows
  never fired.
- Also fixes incipient NOT NULL violations from migration 017.
  That migration added `account_id NOT NULL` to automation_logs,
  automation_pending_executions, flow_runs, and deals. Inserts in
  the engines hadn't been updated yet — any production run would
  500 with a column violation. This PR adds account_id to every
  such insert.

Surface
- `src/app/api/whatsapp/webhook/route.ts`
  - Resolves `config.account_id` AND `config.user_id` from the
    matched whatsapp_config row. Passes accountId through every
    downstream call. user_id is kept as `configOwnerUserId` and
    used only for NOT NULL FK audit columns on contact /
    conversation inserts.
  - `findOrCreateContact` / `findOrCreateConversation` look up by
    account_id and insert with both columns.
  - `flagBroadcastReplyIfAny` matches broadcasts by account_id so
    any teammate's broadcast is updated when a customer replies.
  - `dispatchInboundToFlows` and `runAutomationsForTrigger` now
    take `accountId` instead of `userId`.
- `src/lib/automations/engine.ts`
  - `DispatchInput.userId` → `accountId`.
  - `runAutomationsForTrigger` queries automations by account_id.
  - `executeAutomation` inserts into automation_logs with both
    account_id (tenancy) and user_id (audit, from automation row).
  - `wait` step inserts into automation_pending_executions with
    account_id.
  - `create_deal` step inserts into deals with account_id.
  - `assign_conversation` / `close_conversation` /
    `resolveConversationId` filter by account_id.
  - `resumePendingExecution` row shape now includes account_id;
    cron passes it through.
  - engineSendText / engineSendTemplate calls pass
    `accountId: args.automation.account_id`.
- `src/lib/flows/engine.ts`
  - `DispatchInboundInput.userId` is kept (audit) and
    `accountId` is added (tenancy).
  - `loadActiveRunForContact`, `isDuplicateInbound`,
    `findEntryFlow` all query by account_id.
  - `startNewRun` inserts into flow_runs with account_id.
  - All `engineSend*` calls pass `accountId: run.account_id`.
- `src/lib/automations/meta-send.ts` + `src/lib/flows/meta-send.ts`
  - All four engine senders (text, template, media, interactive
    buttons/list) accept `accountId` and look up the contact +
    whatsapp_config by account_id. user_id stays for inserts.
- `src/lib/flows/types.ts`
  - `FlowRow`, `FlowRunRow`, `DispatchInboundInput` gain
    `account_id` / `accountId`.
- `src/types/index.ts`
  - `Automation` gains `account_id`. Doc note on the user_id
    field clarifying it's audit-only post-multi-user.
- `src/app/api/automations/cron/route.ts`
  - Pass account_id from the pending row to resumePendingExecution.
- `src/app/api/automations/engine/route.ts`
  - Uses `getCurrentAccount()` from #169 to resolve the caller's
    account, then dispatches with accountId.
- `CHANGELOG.md` — Fixed entry explaining the inbox / NOT NULL fix.

Backwards compatibility
- All affected tables already have account_id from migration 017.
  This PR populates it on insert paths that hadn't been updated.
- The `user_id` columns are kept everywhere — they still identify
  the agent / author for audit display. They just no longer
  enforce tenancy.

Verification
- `npm run lint`      → 0 errors (same 19 pre-existing warnings)
- `npm run typecheck` → clean (had 4 errors mid-refactor; all
                        resolved by the call-site updates)
- `npm test`          → 366/366 pass (engine tests are pure unit
                        tests over helper functions; not affected)
- `npm run build`     → succeeds

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Arnas Donauskas 2026-05-31 18:45:48 +03:00
parent fe33bdd40c
commit 0ed07ee6c9
No known key found for this signature in database
10 changed files with 216 additions and 88 deletions

View File

@ -38,6 +38,20 @@ as the sole owner of their own account and sees identical data.
- **The signup trigger (`handle_new_user`) now also creates a - **The signup trigger (`handle_new_user`) now also creates a
personal account** and links the new profile to it as `owner`. personal account** and links the new profile to it as `owner`.
### Fixed
- **Inbound WhatsApp messages now land in the shared inbox.** The
webhook + automations + flows engines used to route inbound
events by `user_id`, which after the 017 migration only matched
the WhatsApp config owner's automations / flows — teammates'
rules never fired. PR 8 of the multi-user series flips every
lookup to `account_id` so any member of the account sees the
inbound message and any teammate's automation or flow can react
to it. Also fixes incipient NOT NULL violations on
`automation_logs`, `automation_pending_executions`, `flow_runs`,
and `deals` — those tables gained `account_id NOT NULL` in 017
but the engines hadn't yet been updated to populate it.
### Added ### Added
- **Account & member management API** — server-side endpoints - **Account & member management API** — server-side endpoints

View File

@ -50,6 +50,9 @@ export async function GET(request: Request) {
await resumePendingExecution({ await resumePendingExecution({
id: row.id as string, id: row.id as string,
automation_id: row.automation_id as string, automation_id: row.automation_id as string,
// account_id is NOT NULL on automation_pending_executions
// post-017; the engine uses it for tenant-scoped lookups.
account_id: row.account_id as string,
user_id: row.user_id as string, user_id: row.user_id as string,
contact_id: (row.contact_id as string | null) ?? null, contact_id: (row.contact_id as string | null) ?? null,
log_id: (row.log_id as string | null) ?? null, log_id: (row.log_id as string | null) ?? null,

View File

@ -1,19 +1,21 @@
import { NextResponse } from 'next/server' import { NextResponse } from 'next/server'
import { createClient } from '@/lib/supabase/server' import { getCurrentAccount, toErrorResponse } from '@/lib/auth/account'
import { runAutomationsForTrigger } from '@/lib/automations/engine' import { runAutomationsForTrigger } from '@/lib/automations/engine'
import type { AutomationTriggerType } from '@/types' import type { AutomationTriggerType } from '@/types'
/** /**
* Manual trigger for testing or for external integrations that want * Manual trigger for testing or for external integrations that want
* to fire automations. Auth is required — the caller's user_id is * to fire automations. Auth is required — we resolve the caller's
* used so RLS-safe data remains per-user. * account_id and dispatch over the account's automations.
*/ */
export async function POST(request: Request) { export async function POST(request: Request) {
const supabase = await createClient() let accountId: string
const { try {
data: { user }, const ctx = await getCurrentAccount()
} = await supabase.auth.getUser() accountId = ctx.accountId
if (!user) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }) } catch (err) {
return toErrorResponse(err)
}
const body = await request.json().catch(() => null) const body = await request.json().catch(() => null)
if (!body?.trigger_type) { if (!body?.trigger_type) {
@ -21,7 +23,7 @@ export async function POST(request: Request) {
} }
await runAutomationsForTrigger({ await runAutomationsForTrigger({
userId: user.id, accountId,
triggerType: body.trigger_type as AutomationTriggerType, triggerType: body.trigger_type as AutomationTriggerType,
contactId: body.contact_id ?? null, contactId: body.contact_id ?? null,
context: body.context ?? {}, context: body.context ?? {},

View File

@ -249,9 +249,9 @@ async function processWebhook(body: { entry?: WhatsAppWebhookEntry[] }) {
console.error( console.error(
`Multiple configs (${configRows.length}) found for phone_number_id:`, `Multiple configs (${configRows.length}) found for phone_number_id:`,
phoneNumberId, phoneNumberId,
'— inbound message dropped. Resolve duplicates so each number maps to a single user.', '— inbound message dropped. Resolve duplicates so each number maps to a single account.',
'Owners:', 'Account owners:',
configRows.map((r: { user_id: string }) => r.user_id) configRows.map((r: { account_id: string; user_id: string }) => `${r.account_id} (admin ${r.user_id})`)
) )
continue continue
} }
@ -267,6 +267,12 @@ async function processWebhook(body: { entry?: WhatsAppWebhookEntry[] }) {
await processMessage( await processMessage(
message, message,
contact, contact,
// Tenancy — drives every contact / conversation lookup
// and the engines' active-row dispatch.
config.account_id,
// Audit / sender-of-record — used as the user_id on row
// inserts that need it for NOT NULL FK compliance. Always
// the admin who saved the WhatsApp config.
config.user_id, config.user_id,
decryptedAccessToken decryptedAccessToken
) )
@ -379,14 +385,17 @@ async function handleStatusUpdate(status: {
* Runs on a best-effort basis — failures here must not break the * Runs on a best-effort basis — failures here must not break the
* main inbound-message flow, so errors are swallowed with a log. * main inbound-message flow, so errors are swallowed with a log.
*/ */
async function flagBroadcastReplyIfAny(userId: string, contactId: string) { async function flagBroadcastReplyIfAny(accountId: string, contactId: string) {
try { try {
// Most recent outbound broadcast that hasn't been replied to yet. // Most recent outbound broadcast in this account that hasn't
// been replied to yet. Account-scoped so a shared inbox reply
// marks the broadcast as replied regardless of which teammate
// sent it.
const { data: recs, error } = await supabaseAdmin() const { data: recs, error } = await supabaseAdmin()
.from('broadcast_recipients') .from('broadcast_recipients')
.select('id, status, broadcast_id, broadcasts!inner(user_id)') .select('id, status, broadcast_id, broadcasts!inner(account_id)')
.eq('contact_id', contactId) .eq('contact_id', contactId)
.eq('broadcasts.user_id', userId) .eq('broadcasts.account_id', accountId)
.in('status', ['sent', 'delivered', 'read']) .in('status', ['sent', 'delivered', 'read'])
.order('created_at', { ascending: false }) .order('created_at', { ascending: false })
.limit(1) .limit(1)
@ -491,7 +500,14 @@ async function handleReaction(
async function processMessage( async function processMessage(
message: WhatsAppMessage, message: WhatsAppMessage,
contact: { profile: { name: string }; wa_id: string }, contact: { profile: { name: string }; wa_id: string },
userId: string, // Tenancy. Resolved from the matched whatsapp_config row; every
// contact / conversation / message row created downstream is
// stamped with this so any member of the account can see it.
accountId: string,
// Sender-of-record for inserts that need a NOT NULL user_id FK
// (contacts, conversations). Always the admin who saved the
// WhatsApp config; the choice is arbitrary post-017 but stable.
configOwnerUserId: string,
accessToken: string accessToken: string
) { ) {
const senderPhone = normalizePhone(message.from) const senderPhone = normalizePhone(message.from)
@ -499,7 +515,8 @@ async function processMessage(
// Find or create contact // Find or create contact
const contactOutcome = await findOrCreateContact( const contactOutcome = await findOrCreateContact(
userId, accountId,
configOwnerUserId,
senderPhone, senderPhone,
contactName contactName
) )
@ -508,7 +525,8 @@ async function processMessage(
// Find or create conversation // Find or create conversation
const conversation = await findOrCreateConversation( const conversation = await findOrCreateConversation(
userId, accountId,
configOwnerUserId,
contactRecord.id contactRecord.id
) )
if (!conversation) return if (!conversation) return
@ -615,7 +633,7 @@ async function processMessage(
// If this contact was a recent broadcast recipient, flag the reply // If this contact was a recent broadcast recipient, flag the reply
// so the broadcast's `replied_count` advances (via the aggregate // so the broadcast's `replied_count` advances (via the aggregate
// trigger installed in migration 003). // trigger installed in migration 003).
await flagBroadcastReplyIfAny(userId, contactRecord.id) await flagBroadcastReplyIfAny(accountId, contactRecord.id)
// ============================================================ // ============================================================
// Flow runner dispatch. // Flow runner dispatch.
@ -637,7 +655,8 @@ async function processMessage(
// basically for free (one indexed SELECT for the active run). // basically for free (one indexed SELECT for the active run).
// ============================================================ // ============================================================
const flowResult = await dispatchInboundToFlows({ const flowResult = await dispatchInboundToFlows({
userId, accountId,
userId: configOwnerUserId,
contactId: contactRecord.id, contactId: contactRecord.id,
conversationId: conversation.id, conversationId: conversation.id,
message: message:
@ -684,7 +703,7 @@ async function processMessage(
if (isFirstInboundMessage) automationTriggers.unshift('first_inbound_message') if (isFirstInboundMessage) automationTriggers.unshift('first_inbound_message')
for (const triggerType of automationTriggers) { for (const triggerType of automationTriggers) {
runAutomationsForTrigger({ runAutomationsForTrigger({
userId, accountId,
triggerType, triggerType,
contactId: contactRecord.id, contactId: contactRecord.id,
context: { context: {
@ -851,15 +870,19 @@ interface ContactOutcome {
} }
async function findOrCreateContact( async function findOrCreateContact(
userId: string, accountId: string,
configOwnerUserId: string,
phone: string, phone: string,
name: string name: string
): Promise<ContactOutcome | null> { ): Promise<ContactOutcome | null> {
// Look up existing contacts for this user // Look up existing contacts for this account. Switched from
// user_id to account_id in PR 8 of the multi-user series so an
// inbound message lands on the team's existing contact row even
// if a different teammate created it.
const { data: contacts, error: contactsError } = await supabaseAdmin() const { data: contacts, error: contactsError } = await supabaseAdmin()
.from('contacts') .from('contacts')
.select('*') .select('*')
.eq('user_id', userId) .eq('account_id', accountId)
if (contactsError) { if (contactsError) {
console.error('Error fetching contacts:', contactsError) console.error('Error fetching contacts:', contactsError)
@ -880,11 +903,15 @@ async function findOrCreateContact(
return { contact: existingContact, wasCreated: false } return { contact: existingContact, wasCreated: false }
} }
// Create new contact // Create new contact. account_id is the tenancy column;
// user_id is the NOT NULL FK audit column (no inbound message
// has a single "user who created" it — we attribute to the
// WhatsApp config owner as a stable default).
const { data: newContact, error: createError } = await supabaseAdmin() const { data: newContact, error: createError } = await supabaseAdmin()
.from('contacts') .from('contacts')
.insert({ .insert({
user_id: userId, account_id: accountId,
user_id: configOwnerUserId,
phone, phone,
name: name || phone, name: name || phone,
}) })
@ -899,12 +926,16 @@ async function findOrCreateContact(
return { contact: newContact, wasCreated: true } return { contact: newContact, wasCreated: true }
} }
async function findOrCreateConversation(userId: string, contactId: string) { async function findOrCreateConversation(
// Look for existing conversation accountId: string,
configOwnerUserId: string,
contactId: string,
) {
// Look for existing conversation in this account
const { data: existing, error: findError } = await supabaseAdmin() const { data: existing, error: findError } = await supabaseAdmin()
.from('conversations') .from('conversations')
.select('*') .select('*')
.eq('user_id', userId) .eq('account_id', accountId)
.eq('contact_id', contactId) .eq('contact_id', contactId)
.single() .single()
@ -912,11 +943,13 @@ async function findOrCreateConversation(userId: string, contactId: string) {
return existing return existing
} }
// Create new conversation // Create new conversation. Same tenancy + audit split as
// findOrCreateContact above.
const { data: newConv, error: createError } = await supabaseAdmin() const { data: newConv, error: createError } = await supabaseAdmin()
.from('conversations') .from('conversations')
.insert({ .insert({
user_id: userId, account_id: accountId,
user_id: configOwnerUserId,
contact_id: contactId, contact_id: contactId,
}) })
.select() .select()

View File

@ -35,14 +35,20 @@ export interface AutomationContext {
} }
export interface DispatchInput { export interface DispatchInput {
userId: string /** Account-level tenancy key. Drives the lookup of which active
* automations to fire — `automations.account_id` is the tenant
* isolation after migration 017. Replaces the previous `userId`
* field; the per-automation user_id is read off each row when
* needed (sender identity for outbound messages, log audit). */
accountId: string
triggerType: AutomationTriggerType triggerType: AutomationTriggerType
contactId?: string | null contactId?: string | null
context?: AutomationContext context?: AutomationContext
} }
/** /**
* Fire all active automations matching the given trigger for a user. * Fire all active automations matching the given trigger for an
* account.
* *
* Must never throw — callers use fire-and-forget from the webhook. * Must never throw — callers use fire-and-forget from the webhook.
* All errors are caught and logged; per-automation failures are * All errors are caught and logged; per-automation failures are
@ -54,7 +60,7 @@ export async function runAutomationsForTrigger(input: DispatchInput): Promise<vo
const { data: automations, error } = await db const { data: automations, error } = await db
.from('automations') .from('automations')
.select('*') .select('*')
.eq('user_id', input.userId) .eq('account_id', input.accountId)
.eq('trigger_type', input.triggerType) .eq('trigger_type', input.triggerType)
.eq('is_active', true) .eq('is_active', true)
@ -84,7 +90,12 @@ export async function runAutomationsForTrigger(input: DispatchInput): Promise<vo
export async function resumePendingExecution(pending: { export async function resumePendingExecution(pending: {
id: string id: string
automation_id: string automation_id: string
/** Audit-only; the automation row carries account_id for tenancy. */
user_id: string user_id: string
/** Account-scoped lookups read from the automation row, so this
* field is just here to mirror the row shape and keep the cron's
* pass-through self-documenting. */
account_id: string
contact_id: string | null contact_id: string | null
log_id: string | null log_id: string | null
parent_step_id: string | null parent_step_id: string | null
@ -134,6 +145,11 @@ async function executeAutomation(automation: Automation, input: DispatchInput) {
.from('automation_logs') .from('automation_logs')
.insert({ .insert({
automation_id: automation.id, automation_id: automation.id,
// Tenancy: matches automation.account_id (NOT NULL post-017).
account_id: automation.account_id,
// Audit: keeps the historical "author of this automation"
// pointer so logs still attribute to the right user even
// after teammates join the account.
user_id: automation.user_id, user_id: automation.user_id,
contact_id: input.contactId ?? null, contact_id: input.contactId ?? null,
trigger_event: input.triggerType, trigger_event: input.triggerType,
@ -222,6 +238,8 @@ async function executeStepsFrom(args: ExecuteArgs): Promise<void> {
const ms = waitMs(cfg) const ms = waitMs(cfg)
await db.from('automation_pending_executions').insert({ await db.from('automation_pending_executions').insert({
automation_id: args.automation.id, automation_id: args.automation.id,
// Tenancy: account_id required NOT NULL post-017.
account_id: args.automation.account_id,
user_id: args.automation.user_id, user_id: args.automation.user_id,
contact_id: args.contactId, contact_id: args.contactId,
log_id: args.logId, log_id: args.logId,
@ -305,6 +323,7 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
if (!text.trim()) throw new Error('send_message has empty text') if (!text.trim()) throw new Error('send_message has empty text')
const conversationId = await resolveConversationId(args) const conversationId = await resolveConversationId(args)
const { whatsapp_message_id } = await engineSendText({ const { whatsapp_message_id } = await engineSendText({
accountId: args.automation.account_id,
userId: args.automation.user_id, userId: args.automation.user_id,
conversationId, conversationId,
contactId: args.contactId, contactId: args.contactId,
@ -337,6 +356,7 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
.map((k) => String(cfg.variables![k])) .map((k) => String(cfg.variables![k]))
: [] : []
const { whatsapp_message_id } = await engineSendTemplate({ const { whatsapp_message_id } = await engineSendTemplate({
accountId: args.automation.account_id,
userId: args.automation.user_id, userId: args.automation.user_id,
conversationId, conversationId,
contactId: args.contactId, contactId: args.contactId,
@ -375,10 +395,13 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
if (!args.contactId) throw new Error('assign_conversation needs a contact') if (!args.contactId) throw new Error('assign_conversation needs a contact')
let agentId = cfg.agent_id let agentId = cfg.agent_id
if (cfg.mode === 'round_robin') { if (cfg.mode === 'round_robin') {
// Pick any member of the account. The existing implementation
// only ever returned the automation's author; preserving that
// shape until a real round-robin algorithm replaces it.
const { data: profiles } = await db const { data: profiles } = await db
.from('profiles') .from('profiles')
.select('user_id') .select('user_id')
.eq('user_id', args.automation.user_id) .eq('account_id', args.automation.account_id)
.limit(1) .limit(1)
agentId = profiles?.[0]?.user_id agentId = profiles?.[0]?.user_id
} }
@ -386,7 +409,7 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
await db await db
.from('conversations') .from('conversations')
.update({ assigned_agent_id: agentId }) .update({ assigned_agent_id: agentId })
.eq('user_id', args.automation.user_id) .eq('account_id', args.automation.account_id)
.eq('contact_id', args.contactId) .eq('contact_id', args.contactId)
return `assigned to ${agentId}` return `assigned to ${agentId}`
} }
@ -409,6 +432,8 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
const cfg = step.step_config as CreateDealStepConfig const cfg = step.step_config as CreateDealStepConfig
if (!cfg.pipeline_id || !cfg.stage_id) throw new Error('create_deal needs pipeline + stage') if (!cfg.pipeline_id || !cfg.stage_id) throw new Error('create_deal needs pipeline + stage')
await db.from('deals').insert({ await db.from('deals').insert({
// Tenancy + audit, same split as automation_logs above.
account_id: args.automation.account_id,
user_id: args.automation.user_id, user_id: args.automation.user_id,
pipeline_id: cfg.pipeline_id, pipeline_id: cfg.pipeline_id,
stage_id: cfg.stage_id, stage_id: cfg.stage_id,
@ -438,7 +463,7 @@ async function runStep(step: AutomationStep, args: ExecuteArgs): Promise<string>
await db await db
.from('conversations') .from('conversations')
.update({ status: 'closed', updated_at: new Date().toISOString() }) .update({ status: 'closed', updated_at: new Date().toISOString() })
.eq('user_id', args.automation.user_id) .eq('account_id', args.automation.account_id)
.eq('contact_id', args.contactId) .eq('contact_id', args.contactId)
return 'conversation closed' return 'conversation closed'
} }
@ -466,7 +491,7 @@ async function resolveConversationId(args: ExecuteArgs): Promise<string> {
const { data, error } = await supabaseAdmin() const { data, error } = await supabaseAdmin()
.from('conversations') .from('conversations')
.select('id') .select('id')
.eq('user_id', args.automation.user_id) .eq('account_id', args.automation.account_id)
.eq('contact_id', args.contactId) .eq('contact_id', args.contactId)
.maybeSingle() .maybeSingle()
if (error) throw new Error(`conversation lookup failed: ${error.message}`) if (error) throw new Error(`conversation lookup failed: ${error.message}`)

View File

@ -20,6 +20,13 @@ import { supabaseAdmin } from './admin-client'
// ------------------------------------------------------------ // ------------------------------------------------------------
interface SendTextArgs { interface SendTextArgs {
/** Account-level tenancy key. Drives contact + whatsapp_config
* lookups so an automation authored by user A still sends through
* the WhatsApp number user B saved on the same account. */
accountId: string
/** Original author of the automation/flow — used for INSERT audit
* columns (messages.sender_id-ish) and for resolving the agent's
* identity in logs. Not consulted for tenancy. */
userId: string userId: string
conversationId: string conversationId: string
contactId: string contactId: string
@ -27,6 +34,7 @@ interface SendTextArgs {
} }
interface SendTemplateArgs { interface SendTemplateArgs {
accountId: string
userId: string userId: string
conversationId: string conversationId: string
contactId: string contactId: string
@ -52,22 +60,22 @@ type SendInput =
async function sendViaMeta(input: SendInput): Promise<{ whatsapp_message_id: string }> { async function sendViaMeta(input: SendInput): Promise<{ whatsapp_message_id: string }> {
const db = supabaseAdmin() const db = supabaseAdmin()
// Scope the contact lookup by user_id. The engine uses the // Scope the contact + config lookups by account_id, not user_id.
// service-role client (bypassing RLS), and the public // The engine uses the service-role client (bypassing RLS); without
// /api/automations/engine endpoint accepts contact_id from the // this filter, an authenticated user could fire their own
// request body — without this filter, an authenticated user could // automations against another tenant's contact UUID and send via
// fire their own automations against another tenant's contact UUID // their own WhatsApp config to that contact's phone. The 017
// and send via their own WhatsApp config to that contact's phone. // migration moved both tables to account-scoped tenancy, so the
// Practical risk is low (UUIDs are unguessable) but the check is // check is the same defense-in-depth as before, just keyed on the
// cheap defense-in-depth. // new tenancy column.
const { data: contact, error: contactErr } = await db const { data: contact, error: contactErr } = await db
.from('contacts') .from('contacts')
.select('id, phone') .select('id, phone')
.eq('id', input.contactId) .eq('id', input.contactId)
.eq('user_id', input.userId) .eq('account_id', input.accountId)
.maybeSingle() .maybeSingle()
if (contactErr || !contact?.phone) { if (contactErr || !contact?.phone) {
throw new Error('contact not found for this user') throw new Error('contact not found for this account')
} }
const sanitized = sanitizePhoneForMeta(contact.phone) const sanitized = sanitizePhoneForMeta(contact.phone)
@ -78,7 +86,7 @@ async function sendViaMeta(input: SendInput): Promise<{ whatsapp_message_id: str
const { data: config, error: configErr } = await db const { data: config, error: configErr } = await db
.from('whatsapp_config') .from('whatsapp_config')
.select('*') .select('*')
.eq('user_id', input.userId) .eq('account_id', input.accountId)
.single() .single()
if (configErr || !config) { if (configErr || !config) {
throw new Error('WhatsApp not configured for this account') throw new Error('WhatsApp not configured for this account')

View File

@ -173,19 +173,21 @@ type AdminClient = ReturnType<typeof supabaseAdmin>;
async function loadActiveRunForContact( async function loadActiveRunForContact(
db: AdminClient, db: AdminClient,
userId: string, accountId: string,
contactId: string, contactId: string,
): Promise<FlowRunRow | null> { ): Promise<FlowRunRow | null> {
// The partial unique index `idx_one_active_run_per_contact` makes // The partial unique index `idx_one_active_run_per_contact` was
// "two active runs for one contact" impossible by design. But a // rebuilt in migration 017 over `(account_id, contact_id)` — so
// future migration glitch or manual SQL could create one, and // "two active runs for one contact in one account" is impossible
// .maybeSingle() throws on >1 row — which would kill dispatch for // by design. But a future migration glitch or manual SQL could
// that contact's webhook entirely. .limit(1) is forgiving: pick the // create one, and .maybeSingle() throws on >1 row — which would
// newest, let the cron sweep clean up the stale one. // kill dispatch for that contact's webhook entirely. .limit(1) is
// forgiving: pick the newest, let the cron sweep clean up the
// stale one.
const { data, error } = await db const { data, error } = await db
.from("flow_runs") .from("flow_runs")
.select("*") .select("*")
.eq("user_id", userId) .eq("account_id", accountId)
.eq("contact_id", contactId) .eq("contact_id", contactId)
.eq("status", "active") .eq("status", "active")
.order("started_at", { ascending: false }) .order("started_at", { ascending: false })
@ -282,16 +284,17 @@ async function logEvent(
*/ */
async function isDuplicateInbound( async function isDuplicateInbound(
db: AdminClient, db: AdminClient,
userId: string, accountId: string,
contactId: string, contactId: string,
metaMessageId: string, metaMessageId: string,
): Promise<boolean> { ): Promise<boolean> {
// Fetch ALL run ids for this contact (active + historical). Bounded // Fetch ALL run ids for this contact in this account (active +
// by how many flows the customer has been through — small. // historical). Bounded by how many flows the customer has been
// through — small.
const { data: runs } = await db const { data: runs } = await db
.from("flow_runs") .from("flow_runs")
.select("id") .select("id")
.eq("user_id", userId) .eq("account_id", accountId)
.eq("contact_id", contactId); .eq("contact_id", contactId);
if (!runs?.length) return false; if (!runs?.length) return false;
const runIds = runs.map((r) => (r as { id: string }).id); const runIds = runs.map((r) => (r as { id: string }).id);
@ -307,7 +310,7 @@ async function isDuplicateInbound(
async function findEntryFlow( async function findEntryFlow(
db: AdminClient, db: AdminClient,
userId: string, accountId: string,
message: ParsedInbound, message: ParsedInbound,
isFirstInbound: boolean, isFirstInbound: boolean,
): Promise<FlowRow | null> { ): Promise<FlowRow | null> {
@ -315,13 +318,13 @@ async function findEntryFlow(
// are responses to existing prompts; they never start a new flow. // are responses to existing prompts; they never start a new flow.
if (message.kind !== "text") return null; if (message.kind !== "text") return null;
// Pull all active flows for this user. Active set is bounded (the // Pull all active flows for this account. Active set is bounded
// builder discourages double-trigger overlap; partial index makes // (the builder discourages double-trigger overlap; partial index
// the lookup index-supported). // makes the lookup index-supported).
const { data: flows, error } = await db const { data: flows, error } = await db
.from("flows") .from("flows")
.select("*") .select("*")
.eq("user_id", userId) .eq("account_id", accountId)
.eq("status", "active") .eq("status", "active")
.order("created_at", { ascending: true }); .order("created_at", { ascending: true });
if (error || !flows) return null; if (error || !flows) return null;
@ -356,6 +359,7 @@ async function sendButtonsAndSuspend(
): Promise<{ outcome: "advanced"; node_key: string }> { ): Promise<{ outcome: "advanced"; node_key: string }> {
const cfg = node.config as unknown as SendButtonsNodeConfig; const cfg = node.config as unknown as SendButtonsNodeConfig;
const { whatsapp_message_id } = await engineSendInteractiveButtons({ const { whatsapp_message_id } = await engineSendInteractiveButtons({
accountId: run.account_id,
userId: run.user_id, userId: run.user_id,
conversationId: run.conversation_id!, conversationId: run.conversation_id!,
contactId: run.contact_id!, contactId: run.contact_id!,
@ -391,6 +395,7 @@ async function sendListAndSuspend(
): Promise<{ outcome: "advanced"; node_key: string }> { ): Promise<{ outcome: "advanced"; node_key: string }> {
const cfg = node.config as unknown as SendListNodeConfig; const cfg = node.config as unknown as SendListNodeConfig;
const { whatsapp_message_id } = await engineSendInteractiveList({ const { whatsapp_message_id } = await engineSendInteractiveList({
accountId: run.account_id,
userId: run.user_id, userId: run.user_id,
conversationId: run.conversation_id!, conversationId: run.conversation_id!,
contactId: run.contact_id!, contactId: run.contact_id!,
@ -576,7 +581,8 @@ async function advanceFromNodeKey(
const cfg = node.config as unknown as SendMessageNodeConfig; const cfg = node.config as unknown as SendMessageNodeConfig;
try { try {
const { whatsapp_message_id } = await engineSendText({ const { whatsapp_message_id } = await engineSendText({
userId: run.user_id, accountId: run.account_id,
userId: run.user_id,
conversationId: run.conversation_id!, conversationId: run.conversation_id!,
contactId: run.contact_id!, contactId: run.contact_id!,
text: interpolateVars(cfg.text, run.vars), text: interpolateVars(cfg.text, run.vars),
@ -600,7 +606,8 @@ async function advanceFromNodeKey(
const cfg = node.config as unknown as SendMediaNodeConfig; const cfg = node.config as unknown as SendMediaNodeConfig;
try { try {
const { whatsapp_message_id } = await engineSendMedia({ const { whatsapp_message_id } = await engineSendMedia({
userId: run.user_id, accountId: run.account_id,
userId: run.user_id,
conversationId: run.conversation_id!, conversationId: run.conversation_id!,
contactId: run.contact_id!, contactId: run.contact_id!,
kind: cfg.media_type, kind: cfg.media_type,
@ -632,7 +639,8 @@ async function advanceFromNodeKey(
const cfg = node.config as unknown as CollectInputNodeConfig; const cfg = node.config as unknown as CollectInputNodeConfig;
try { try {
const { whatsapp_message_id } = await engineSendText({ const { whatsapp_message_id } = await engineSendText({
userId: run.user_id, accountId: run.account_id,
userId: run.user_id,
conversationId: run.conversation_id!, conversationId: run.conversation_id!,
contactId: run.contact_id!, contactId: run.contact_id!,
text: interpolateVars(cfg.prompt_text, run.vars), text: interpolateVars(cfg.prompt_text, run.vars),
@ -825,7 +833,7 @@ export async function dispatchInboundToFlows(
try { try {
const activeRun = await loadActiveRunForContact( const activeRun = await loadActiveRunForContact(
db, db,
input.userId, input.accountId,
input.contactId, input.contactId,
); );
@ -835,7 +843,7 @@ export async function dispatchInboundToFlows(
if (activeRun) { if (activeRun) {
const dupe = await isDuplicateInbound( const dupe = await isDuplicateInbound(
db, db,
input.userId, input.accountId,
input.contactId, input.contactId,
input.message.meta_message_id, input.message.meta_message_id,
); );
@ -855,7 +863,7 @@ export async function dispatchInboundToFlows(
// No active run → look for a flow whose entry trigger matches. // No active run → look for a flow whose entry trigger matches.
const flow = await findEntryFlow( const flow = await findEntryFlow(
db, db,
input.userId, input.accountId,
input.message, input.message,
input.isFirstInboundMessage, input.isFirstInboundMessage,
); );
@ -1007,7 +1015,8 @@ async function handleReplyForActiveRun(
const cfg = currentNode.config as unknown as CollectInputNodeConfig; const cfg = currentNode.config as unknown as CollectInputNodeConfig;
try { try {
await engineSendText({ await engineSendText({
userId: run.user_id, accountId: run.account_id,
userId: run.user_id,
conversationId: run.conversation_id!, conversationId: run.conversation_id!,
contactId: run.contact_id!, contactId: run.contact_id!,
text: interpolateVars(cfg.prompt_text, run.vars), text: interpolateVars(cfg.prompt_text, run.vars),
@ -1052,6 +1061,13 @@ async function startNewRun(
.from("flow_runs") .from("flow_runs")
.insert({ .insert({
flow_id: flow.id, flow_id: flow.id,
// Tenancy: NOT NULL post-017. The partial unique index
// `idx_one_active_run_per_contact` is over (account_id,
// contact_id) WHERE status='active', so two accounts sharing
// a contact phone number each run their own flows independently.
account_id: flow.account_id,
// Audit: preserves the flow's author on the run row for log
// attribution.
user_id: flow.user_id, user_id: flow.user_id,
contact_id: input.contactId, contact_id: input.contactId,
conversation_id: input.conversationId, conversation_id: input.conversationId,

View File

@ -32,6 +32,13 @@ import { supabaseAdmin } from './admin-client'
// ------------------------------------------------------------ // ------------------------------------------------------------
interface SendTextEngineArgs { interface SendTextEngineArgs {
/** Account-level tenancy key. Drives contact + whatsapp_config
* lookups so a flow authored by user A still sends through the
* WhatsApp number user B saved on the same account. */
accountId: string
/** Original author of the flow — used for INSERT audit columns
* and for resolving the agent's identity in logs. Not consulted
* for tenancy. */
userId: string userId: string
conversationId: string conversationId: string
contactId: string contactId: string
@ -59,10 +66,10 @@ export async function engineSendText(
.from('contacts') .from('contacts')
.select('id, phone') .select('id, phone')
.eq('id', args.contactId) .eq('id', args.contactId)
.eq('user_id', args.userId) .eq('account_id', args.accountId)
.maybeSingle() .maybeSingle()
if (contactErr || !contact?.phone) { if (contactErr || !contact?.phone) {
throw new Error('contact not found for this user') throw new Error('contact not found for this account')
} }
const sanitized = sanitizePhoneForMeta(contact.phone) const sanitized = sanitizePhoneForMeta(contact.phone)
@ -73,7 +80,7 @@ export async function engineSendText(
const { data: config, error: configErr } = await db const { data: config, error: configErr } = await db
.from('whatsapp_config') .from('whatsapp_config')
.select('*') .select('*')
.eq('user_id', args.userId) .eq('account_id', args.accountId)
.single() .single()
if (configErr || !config) { if (configErr || !config) {
throw new Error('WhatsApp not configured for this account') throw new Error('WhatsApp not configured for this account')
@ -138,6 +145,7 @@ export async function engineSendText(
} }
interface SendMediaEngineArgs { interface SendMediaEngineArgs {
accountId: string
userId: string userId: string
conversationId: string conversationId: string
contactId: string contactId: string
@ -167,10 +175,10 @@ export async function engineSendMedia(
.from('contacts') .from('contacts')
.select('id, phone') .select('id, phone')
.eq('id', args.contactId) .eq('id', args.contactId)
.eq('user_id', args.userId) .eq('account_id', args.accountId)
.maybeSingle() .maybeSingle()
if (contactErr || !contact?.phone) { if (contactErr || !contact?.phone) {
throw new Error('contact not found for this user') throw new Error('contact not found for this account')
} }
const sanitized = sanitizePhoneForMeta(contact.phone) const sanitized = sanitizePhoneForMeta(contact.phone)
@ -181,7 +189,7 @@ export async function engineSendMedia(
const { data: config, error: configErr } = await db const { data: config, error: configErr } = await db
.from('whatsapp_config') .from('whatsapp_config')
.select('*') .select('*')
.eq('user_id', args.userId) .eq('account_id', args.accountId)
.single() .single()
if (configErr || !config) { if (configErr || !config) {
throw new Error('WhatsApp not configured for this account') throw new Error('WhatsApp not configured for this account')
@ -254,6 +262,7 @@ export async function engineSendMedia(
} }
interface SendInteractiveButtonsEngineArgs { interface SendInteractiveButtonsEngineArgs {
accountId: string
userId: string userId: string
conversationId: string conversationId: string
contactId: string contactId: string
@ -264,6 +273,7 @@ interface SendInteractiveButtonsEngineArgs {
} }
interface SendInteractiveListEngineArgs { interface SendInteractiveListEngineArgs {
accountId: string
userId: string userId: string
conversationId: string conversationId: string
contactId: string contactId: string
@ -310,19 +320,17 @@ async function sendInteractiveViaMeta(
): Promise<{ whatsapp_message_id: string }> { ): Promise<{ whatsapp_message_id: string }> {
const db = supabaseAdmin() const db = supabaseAdmin()
// Scope the contact lookup by user_id — same defense-in-depth // Scope the contact + whatsapp_config lookups by account_id —
// rationale as automations/meta-send.ts. Service-role client // same defense-in-depth rationale as automations/meta-send.ts.
// bypasses RLS, so an attacker who could call into the engine // Migration 017 moved both tables to account-scoped tenancy.
// with a contact_id from another tenant would otherwise send
// through their own WhatsApp config to a stranger's number.
const { data: contact, error: contactErr } = await db const { data: contact, error: contactErr } = await db
.from('contacts') .from('contacts')
.select('id, phone') .select('id, phone')
.eq('id', input.contactId) .eq('id', input.contactId)
.eq('user_id', input.userId) .eq('account_id', input.accountId)
.maybeSingle() .maybeSingle()
if (contactErr || !contact?.phone) { if (contactErr || !contact?.phone) {
throw new Error('contact not found for this user') throw new Error('contact not found for this account')
} }
const sanitized = sanitizePhoneForMeta(contact.phone) const sanitized = sanitizePhoneForMeta(contact.phone)
@ -333,7 +341,7 @@ async function sendInteractiveViaMeta(
const { data: config, error: configErr } = await db const { data: config, error: configErr } = await db
.from('whatsapp_config') .from('whatsapp_config')
.select('*') .select('*')
.eq('user_id', input.userId) .eq('account_id', input.accountId)
.single() .single()
if (configErr || !config) { if (configErr || !config) {
throw new Error('WhatsApp not configured for this account') throw new Error('WhatsApp not configured for this account')

View File

@ -225,6 +225,11 @@ export type FlowTriggerConfig =
export interface FlowRow { export interface FlowRow {
id: string; id: string;
/** Account tenancy (NOT NULL post-017). The engine looks up active
* flows for inbound dispatch using this field. */
account_id: string;
/** Author. Used as a default sender-of-record on engine sends and
* preserved on flow_runs for log/audit display. */
user_id: string; user_id: string;
name: string; name: string;
description: string | null; description: string | null;
@ -253,6 +258,9 @@ export interface FlowNodeRow {
export interface FlowRunRow { export interface FlowRunRow {
id: string; id: string;
flow_id: string; flow_id: string;
/** Tenancy. Matches flows.account_id; NOT NULL post-017. */
account_id: string;
/** Audit. Matches the parent flow.user_id. */
user_id: string; user_id: string;
contact_id: string | null; contact_id: string | null;
conversation_id: string | null; conversation_id: string | null;
@ -322,6 +330,11 @@ export type ParsedInbound =
}; };
export interface DispatchInboundInput { export interface DispatchInboundInput {
/** Account tenancy key. Drives the lookup of active flows and the
* idempotency check for previously-seen inbound message_ids. */
accountId: string;
/** Sender-of-record for the bot's outbound prompts on engine
* sends. Set by the webhook to the WhatsApp config owner. */
userId: string; userId: string;
contactId: string; contactId: string;
conversationId: string; conversationId: string;

View File

@ -480,6 +480,12 @@ export type AutomationStepConfig =
export interface Automation { export interface Automation {
id: string; id: string;
/** Account tenancy key — every automation belongs to one account
* (migration 017 made the column NOT NULL). The engine looks up
* active automations by this field on inbound webhook events. */
account_id: string;
/** Original author. Used for log audit + outbound message
* sender-of-record, never for tenancy isolation. */
user_id: string; user_id: string;
name: string; name: string;
description?: string; description?: string;