diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cc06ec..180de01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ Versions follow [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). Pre-1.0, `MINOR` bumps cover new modules; `PATCH` bumps cover bug fixes and polish. -## [Unreleased] +## [0.3.0] — 2026-07-01 Multi-user accounts ship. Every wacrm install is multi-tenant on the database side: a single user's signup creates a fresh "account", and @@ -32,10 +32,30 @@ always did. `Authorization: Bearer `. Keys are account-scoped and stored hashed (plaintext shown once). This release ships the auth layer, scopes, per-key rate limiting, the management UI, and a - `GET /api/v1/me` probe to verify a key; the data endpoints - (`messages`, `contacts`, …) follow one at a time. See + `GET /api/v1/me` probe to verify a key. See `docs/public-api.md`. **Migration required:** apply `supabase/migrations/026_api_keys.sql`. ([#245](https://github.com/ArnasDon/wacrm/issues/245)) +- **Public REST API — data endpoints.** Built on the key auth above, + so external automations can read and drive the CRM: + - `POST /api/v1/messages` — send a text / template / media message to + a phone number; finds-or-creates the contact + conversation + (`messages:send`). + - `GET/POST /api/v1/contacts`, `GET/PATCH /api/v1/contacts/{id}` — + list (search + tag filter), create (find-or-create by phone), read, + and update contacts, including tags (`contacts:read` / + `contacts:write`). + - `GET /api/v1/conversations`, `GET /api/v1/conversations/{id}`, and + `GET /api/v1/conversations/{id}/messages` — browse conversations and + their message history with delivery status (`conversations:read` / + `messages:read`). + - `POST /api/v1/broadcasts` + `GET /api/v1/broadcasts/{id}` — launch a + template broadcast to a recipient list and poll its progress + (`broadcasts:send`). + All list endpoints share one cursor-pagination contract + (`{ data, meta: { next_cursor } }`). No migration required — the + scopes already existed and the tables are unchanged. Outbound event + webhooks (react to inbound messages) are the remaining roadmap item. + See `docs/public-api.md`. ([#245](https://github.com/ArnasDon/wacrm/issues/245)) ### Changed diff --git a/docs/public-api.md b/docs/public-api.md index c678d5b..0201fc3 100644 --- a/docs/public-api.md +++ b/docs/public-api.md @@ -4,10 +4,10 @@ The public API lets you drive your wacrm instance from your own scripts and automations — send messages, manage contacts, launch broadcasts — without going through the dashboard UI. -> **Status:** groundwork release. Authentication, scopes, rate -> limiting, and the `GET /api/v1/me` probe ship now. The data -> endpoints (`messages`, `contacts`, …) land one at a time in -> follow-up releases — see [Roadmap](#roadmap). +> **Status:** stable. Authentication, scopes, rate limiting, and the +> messages / contacts / conversations / broadcasts endpoints ship now. +> The one remaining roadmap item is outbound event webhooks — see +> [Roadmap](#roadmap). ## Authentication @@ -115,16 +115,181 @@ curl https://your-crm.example.com/api/v1/me \ } ``` +### `POST /api/v1/messages` + +Send a WhatsApp message to a phone number. Scope: `messages:send`. You +pass an **E.164 number**, not an internal id — the endpoint +finds-or-creates the contact + conversation, then sends. + +```bash +curl -X POST https://your-crm.example.com/api/v1/messages \ + -H "Authorization: Bearer wacrm_live_xxx" \ + -H "Content-Type: application/json" \ + -d '{ "to": "+14155550123", "type": "text", "text": "Hi 👋" }' +``` + +`type` is `text` (default), `template`, or a media kind (`image` / +`video` / `document` / `audio`). Media needs `media_url` (and optional +`filename`); `text` doubles as the caption. `template` needs a +`template` object: + +```jsonc +{ + "to": "+14155550123", + "type": "template", + "template": { + "name": "order_update", + "language": "en_US", + "params": ["A123"] // positional body vars, or a structured object + }, + "reply_to_message_id": "" // optional; must be in the same conversation +} +``` + +Response (201): + +```json +{ + "data": { + "message_id": "…", + "whatsapp_message_id": "wamid.…", + "conversation_id": "…", + "contact_id": "…", + "contact_created": true + } +} +``` + +Domain error codes beyond the table above: `whatsapp_not_configured` +(400), `meta_error` (502 — the request reached Meta and it rejected the +send), `template_malformed` (500). + +### `GET /api/v1/contacts` + +List contacts, newest first. Scope: `contacts:read`. Paginated (see +[Pagination](#pagination)). Optional filters: `?search=` (matches name +or phone) and `?tag=`. + +```json +{ + "data": [ + { + "id": "…", "phone": "+14155550123", "name": "Jane Doe", + "email": null, "company": "Acme", "avatar_url": null, + "tags": [{ "id": "…", "name": "vip", "color": "#3b82f6" }], + "created_at": "…", "updated_at": "…" + } + ], + "meta": { "next_cursor": "…" } +} +``` + +### `POST /api/v1/contacts` + +Create a contact. Scope: `contacts:write`. `phone` (E.164) is required; +`name`, `email`, `company`, and `tags` (an array of tag names, created +if missing) are optional. **Find-or-create by phone:** an existing +match returns `200` with the existing contact; a new contact returns +`201`. The response body is the serialized contact (same shape as the +list rows above). + +### `GET` / `PATCH /api/v1/contacts/{id}` + +Read or update one contact. Scopes: `contacts:read` / `contacts:write`. +`PATCH` updates only the fields you send (`name`, `email`, `company`); +pass `tags` (an array of tag names) to replace the contact's tags. A +contact in another account returns `404`. + +### `GET /api/v1/conversations` + +List conversations, newest first. Scope: `conversations:read`. +Paginated. Optional filters: `?status=` (`open` / `pending` / `closed`) +and `?contact_id=`. Each conversation embeds its contact + tags. + +### `GET /api/v1/conversations/{id}` + +Read one conversation. Scope: `conversations:read`. `404` if it belongs +to another account. + +### `GET /api/v1/conversations/{id}/messages` + +List a conversation's messages, newest first. Scope: `messages:read`. +Paginated. Each message includes its `direction` (`inbound` / +`outbound`), `status` (delivery state), `whatsapp_message_id`, and +`content_*`. The conversation is verified to belong to your account +first (`404` otherwise). + +### `POST /api/v1/broadcasts` + +Launch a template broadcast to a list of recipients. Scope: +`broadcasts:send`. The broadcast + its recipient rows are persisted +immediately and the sends fan out in the background, so the call +returns fast — poll `GET /api/v1/broadcasts/{id}` for progress. + +```bash +curl -X POST https://your-crm.example.com/api/v1/broadcasts \ + -H "Authorization: Bearer wacrm_live_xxx" \ + -H "Content-Type: application/json" \ + -d '{ + "name": "July promo", + "template_name": "promo_july", + "template_language": "en_US", + "recipients": [ + { "to": "+14155550123", "params": ["Jane"] }, + { "to": "+14155550124" } + ] + }' +``` + +Recipients are capped at **1000 per request** — split larger sends. +Invalid phone numbers are dropped and counted as `rejected`. Response +(202): + +```json +{ + "data": { + "broadcast_id": "…", + "status": "sending", + "total_recipients": 2, + "accepted": 2, + "rejected": 0 + } +} +``` + +### `GET /api/v1/broadcasts/{id}` + +Broadcast status + counts. Scope: `broadcasts:send`. `status` moves +`sending` → `sent`; `delivered_count` / `read_count` keep climbing as +Meta delivery webhooks arrive. `404` for another account's broadcast. + +## Pagination + +Every list endpoint pages the same way. Request a page size with +`?limit=` (default 50, max 100) and read the next page with the opaque +`meta.next_cursor` from the previous response: + +``` +GET /api/v1/contacts?limit=50 +→ { "data": [ … ], "meta": { "next_cursor": "eyJ…" } } + +GET /api/v1/contacts?limit=50&cursor=eyJ… +→ { "data": [ … ], "meta": { "next_cursor": null } } // last page +``` + +Cursors are keyset-based (stable under concurrent inserts). Pass the +cursor back verbatim — don't parse it. `next_cursor: null` means the +last page. + ## Roadmap -Planned endpoints, shipping one per release (tracked in -[#245](https://github.com/ArnasDon/wacrm/issues/245)): +The one remaining item tracked in +[#245](https://github.com/ArnasDon/wacrm/issues/245): -- `POST /api/v1/messages` — send a message to a phone number - (`messages:send`) -- `GET/POST /api/v1/contacts`, `GET/PATCH /api/v1/contacts/{id}` - (`contacts:read` / `contacts:write`) -- `GET /api/v1/conversations` (`conversations:read`) -- `POST /api/v1/broadcasts` (`broadcasts:send`) -- Outbound event webhooks (so automations can react to inbound - messages) +- **Outbound event webhooks** — register a URL to receive signed POSTs + when things happen in your account (`message.received`, + `message.status_updated`, `conversation.created`) so automations can + *react* to inbound activity instead of polling. Planned: + a `webhook_endpoints` table (next migration, `028_*`), a + `webhooks:manage` scope, `/api/v1/webhooks` management endpoints, and + HMAC-`X-Wacrm-Signature`-signed delivery with retry/backoff. diff --git a/package.json b/package.json index f14d63d..a8ec223 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "wacrm", - "version": "0.2.2", + "version": "0.3.0", "private": true, "description": "Self-hostable CRM template for WhatsApp built on Next.js and Supabase — shared inbox, contacts, sales pipelines, broadcasts, and no-code automations.", "license": "MIT", diff --git a/src/app/api/v1/broadcasts/[id]/route.ts b/src/app/api/v1/broadcasts/[id]/route.ts new file mode 100644 index 0000000..aeb7125 --- /dev/null +++ b/src/app/api/v1/broadcasts/[id]/route.ts @@ -0,0 +1,41 @@ +// ============================================================ +// GET /api/v1/broadcasts/{id} — broadcast status + counts +// (scope: broadcasts:send). +// +// Poll this after POST /api/v1/broadcasts to watch the fan-out +// progress. `status` moves 'sending' → 'sent'; the delivered/read +// counts continue to climb as Meta delivery webhooks arrive. +// Account-scoped: a foreign id → 404. +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { ok, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; + +export async function GET( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const ctx = await requireApiKey(request, 'broadcasts:send'); + const { id } = await params; + + const { data, error } = await ctx.supabase + .from('broadcasts') + .select( + 'id, name, template_name, template_language, status, total_recipients, sent_count, delivered_count, read_count, replied_count, failed_count, created_at, updated_at' + ) + .eq('id', id) + .eq('account_id', ctx.accountId) + .maybeSingle(); + + if (error) { + console.error('[api/v1/broadcasts] read error:', error); + return fail('internal', 'Failed to read broadcast', 500); + } + if (!data) return fail('not_found', 'Broadcast not found', 404); + + return ok(data); + } catch (err) { + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/broadcasts/route.ts b/src/app/api/v1/broadcasts/route.ts new file mode 100644 index 0000000..eaeb632 --- /dev/null +++ b/src/app/api/v1/broadcasts/route.ts @@ -0,0 +1,105 @@ +// ============================================================ +// POST /api/v1/broadcasts — launch a template broadcast +// (scope: broadcasts:send). +// +// Body: +// { +// "name": "July promo", // optional label +// "template_name": "promo_july", // required, approved template +// "template_language": "en_US", // optional (default en_US) +// "recipients": [ // required, 1..1000 +// { "to": "+14155550123", "params": ["Jane"] }, +// { "to": "+14155550124" } +// ] +// } +// +// The broadcast + its recipient rows are persisted synchronously, then +// the Meta fan-out runs in `after()` so the request returns fast. Poll +// `GET /api/v1/broadcasts/{id}` for progress. +// +// Response (202): +// { "data": { "broadcast_id", "status": "sending", +// "total_recipients", "accepted", "rejected" } } +// ============================================================ + +import { after } from 'next/server'; + +import { requireApiKey } from '@/lib/auth/api-context'; + +// The `after()` fan-out below sends to every recipient sequentially and +// runs within this route's max duration (the same constraint the +// webhook route documents). Give it headroom beyond the platform +// default so a modest batch isn't cut off mid-send — which would leave +// recipient rows 'pending' and the broadcast stuck 'sending'. This is a +// bound, not a guarantee: a near-cap (MAX_RECIPIENTS) audience can +// still exceed 60s, so very large sends should be split across +// requests. A durable queue/cron drain is the complete fix (follow-up). +export const maxDuration = 60; +import { ok, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { resolveAuditUserId, ContactError } from '@/lib/api/v1/contacts'; +import { + createBroadcast, + deliverBroadcast, + BroadcastError, +} from '@/lib/whatsapp/broadcast-core'; + +export async function POST(request: Request) { + try { + const ctx = await requireApiKey(request, 'broadcasts:send'); + + const body = (await request.json().catch(() => null)) as Record< + string, + unknown + > | null; + if (!body || typeof body !== 'object') { + return fail('bad_request', 'Request body must be a JSON object', 400); + } + + const templateName = + typeof body.template_name === 'string' ? body.template_name : ''; + const recipients = Array.isArray(body.recipients) ? body.recipients : []; + + const auditUserId = await resolveAuditUserId(ctx.supabase, ctx.accountId); + + const plan = await createBroadcast(ctx.supabase, ctx.accountId, auditUserId, { + name: typeof body.name === 'string' ? body.name : null, + templateName, + templateLanguage: + typeof body.template_language === 'string' + ? body.template_language + : null, + recipients: recipients.map((r) => ({ + to: typeof r?.to === 'string' ? r.to : '', + params: Array.isArray(r?.params) ? r.params : undefined, + })), + }); + + // Fan out after the response is sent. Uses the same service-role + // client — no request-scoped auth needed for the Meta calls or + // the account-scoped row updates. + after(() => deliverBroadcast(ctx.supabase, plan)); + + return ok( + { + broadcast_id: plan.broadcastId, + status: 'sending', + total_recipients: plan.planned.length, + accepted: plan.planned.length, + rejected: plan.rejected, + }, + 202 + ); + } catch (err) { + if (err instanceof BroadcastError) { + return fail(err.code, err.message, err.status); + } + if (err instanceof ContactError) { + return fail( + err.status === 400 ? 'bad_request' : 'internal', + err.message, + err.status + ); + } + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/contacts/[id]/route.ts b/src/app/api/v1/contacts/[id]/route.ts new file mode 100644 index 0000000..58f1fd0 --- /dev/null +++ b/src/app/api/v1/contacts/[id]/route.ts @@ -0,0 +1,102 @@ +// ============================================================ +// GET /api/v1/contacts/{id} — read a contact (scope: contacts:read) +// PATCH /api/v1/contacts/{id} — update a contact (scope: contacts:write) +// +// Both are account-scoped: a contact belonging to another account +// returns 404 (never 403 — don't reveal it exists elsewhere). +// PATCH updates only the fields present in the body; pass `tags` (an +// array of tag names) to replace the contact's tags. +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { ok, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { + getContactById, + setContactTags, + resolveAuditUserId, + ContactError, +} from '@/lib/api/v1/contacts'; + +export async function GET( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const ctx = await requireApiKey(request, 'contacts:read'); + const { id } = await params; + const contact = await getContactById(ctx.supabase, ctx.accountId, id); + if (!contact) return fail('not_found', 'Contact not found', 404); + return ok(contact); + } catch (err) { + return toApiErrorResponse(err); + } +} + +export async function PATCH( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const ctx = await requireApiKey(request, 'contacts:write'); + const { id } = await params; + + const body = (await request.json().catch(() => null)) as Record< + string, + unknown + > | null; + if (!body || typeof body !== 'object') { + return fail('bad_request', 'Request body must be a JSON object', 400); + } + + // Verify the contact is in this account before mutating anything. + const existing = await getContactById(ctx.supabase, ctx.accountId, id); + if (!existing) return fail('not_found', 'Contact not found', 404); + + // Build a partial update from the provided scalar fields. A field + // is updated only when its key is PRESENT (so omitted fields are + // untouched); `null` clears it, a string sets it, and any other + // type is a 400 rather than a silently-ignored no-op. + const updates: Record = {}; + for (const field of ['name', 'email', 'company'] as const) { + if (!(field in body)) continue; + const value = body[field]; + if (value === null || typeof value === 'string') { + updates[field] = value; + } else { + return fail('bad_request', `'${field}' must be a string or null`, 400); + } + } + + if (Object.keys(updates).length > 0) { + updates.updated_at = new Date().toISOString(); + const { error } = await ctx.supabase + .from('contacts') + .update(updates) + .eq('id', id) + .eq('account_id', ctx.accountId); + if (error) { + console.error('[api/v1/contacts] update error:', error); + return fail('internal', 'Failed to update contact', 500); + } + } + + if (Array.isArray(body.tags)) { + const auditUserId = await resolveAuditUserId(ctx.supabase, ctx.accountId); + await setContactTags( + ctx.supabase, + ctx.accountId, + auditUserId, + id, + body.tags.filter((t): t is string => typeof t === 'string') + ); + } + + const contact = await getContactById(ctx.supabase, ctx.accountId, id); + return ok(contact); + } catch (err) { + if (err instanceof ContactError) { + return fail(err.status === 400 ? 'bad_request' : 'internal', err.message, err.status); + } + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/contacts/route.ts b/src/app/api/v1/contacts/route.ts new file mode 100644 index 0000000..e3d048a --- /dev/null +++ b/src/app/api/v1/contacts/route.ts @@ -0,0 +1,149 @@ +// ============================================================ +// GET /api/v1/contacts — list contacts (scope: contacts:read) +// POST /api/v1/contacts — create a contact (scope: contacts:write) +// +// List is keyset-paginated (see src/lib/api/v1/pagination.ts) and +// supports `?search=` (name/phone) and `?tag=` filters. Create +// is find-or-create by phone: an existing match returns 200 with +// `created: false`; a new row returns 201 with `created: true`. +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { ok, okList, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { + parseListParams, + keysetFilter, + buildPage, +} from '@/lib/api/v1/pagination'; +import { + CONTACT_SELECT, + serializeContact, + findOrCreateContact, + setContactTags, + getContactById, + resolveAuditUserId, + ContactError, +} from '@/lib/api/v1/contacts'; + +// PostgREST filter values are comma/paren-delimited; strip anything +// that could break the `.or()` grammar before interpolating a search +// term. Leaves the characters a phone or name legitimately contains. +function sanitizeSearch(raw: string): string { + return raw.replace(/[^\p{L}\p{N} +@.\-_]/gu, '').trim(); +} + +export async function GET(request: Request) { + try { + const ctx = await requireApiKey(request, 'contacts:read'); + const { limit, cursor } = parseListParams(request); + const url = new URL(request.url); + const search = sanitizeSearch(url.searchParams.get('search') ?? ''); + const tag = url.searchParams.get('tag'); + + // When filtering by tag, add an aliased INNER join on contact_tags + // used purely for the WHERE — the parent is kept only if it has the + // tag. The main `contact_tags(tags(*))` embed still returns the + // contact's FULL tag set for serialization. This filters in one + // bounded query (paged by limit+1) instead of pre-fetching an + // unbounded id list into an `.in(...)`. + const selectClause = tag + ? `${CONTACT_SELECT}, tag_filter:contact_tags!inner(tag_id)` + : CONTACT_SELECT; + + let query = ctx.supabase + .from('contacts') + .select(selectClause) + .eq('account_id', ctx.accountId); + + if (search) { + query = query.or(`name.ilike.*${search}*,phone.ilike.*${search}*`); + } + + if (tag) { + query = query.eq('tag_filter.tag_id', tag); + } + + query = query + .order('created_at', { ascending: false }) + .order('id', { ascending: false }) + .limit(limit + 1); + + const kf = keysetFilter(cursor); + if (kf) query = query.or(kf); + + const { data, error } = await query; + if (error) { + console.error('[api/v1/contacts] list error:', error); + return fail('internal', 'Failed to list contacts', 500); + } + + // Cast via unknown: the conditional `selectClause` (with the + // tag_filter alias) is a runtime string, so supabase-js can't infer + // a row type from it. + const { items, nextCursor } = buildPage( + (data ?? []) as unknown as Array<{ created_at: string; id: string }>, + limit + ); + return okList( + items.map((r) => serializeContact(r as Record)), + nextCursor + ); + } catch (err) { + return toApiErrorResponse(err); + } +} + +export async function POST(request: Request) { + try { + const ctx = await requireApiKey(request, 'contacts:write'); + + const body = (await request.json().catch(() => null)) as Record< + string, + unknown + > | null; + if (!body || typeof body !== 'object') { + return fail('bad_request', 'Request body must be a JSON object', 400); + } + + const phone = typeof body.phone === 'string' ? body.phone.trim() : ''; + if (!phone) { + return fail('bad_request', "'phone' is required", 400); + } + + const auditUserId = await resolveAuditUserId(ctx.supabase, ctx.accountId); + + const { id, created } = await findOrCreateContact( + ctx.supabase, + ctx.accountId, + auditUserId, + { + phone, + name: typeof body.name === 'string' ? body.name : undefined, + email: typeof body.email === 'string' ? body.email : undefined, + company: typeof body.company === 'string' ? body.company : undefined, + } + ); + + if (Array.isArray(body.tags)) { + await setContactTags( + ctx.supabase, + ctx.accountId, + auditUserId, + id, + body.tags.filter((t): t is string => typeof t === 'string') + ); + } + + const contact = await getContactById(ctx.supabase, ctx.accountId, id); + return ok(contact, created ? 201 : 200); + } catch (err) { + if (err instanceof ContactError) { + return fail( + err.status === 400 ? 'bad_request' : 'internal', + err.message, + err.status + ); + } + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/conversations/[id]/messages/route.ts b/src/app/api/v1/conversations/[id]/messages/route.ts new file mode 100644 index 0000000..5b29121 --- /dev/null +++ b/src/app/api/v1/conversations/[id]/messages/route.ts @@ -0,0 +1,65 @@ +// ============================================================ +// GET /api/v1/conversations/{id}/messages — list a conversation's +// messages (scope: messages:read), newest first, keyset-paginated. +// +// The conversation is verified to belong to the key's account before +// any message is returned — a foreign or unknown id → 404. +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { okList, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { + parseListParams, + keysetFilter, + buildPage, +} from '@/lib/api/v1/pagination'; +import { serializeMessage } from '@/lib/api/v1/conversations'; +import type { Message } from '@/types'; + +export async function GET( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const ctx = await requireApiKey(request, 'messages:read'); + const { id } = await params; + const { limit, cursor } = parseListParams(request); + + // Gate on account ownership of the conversation first. + const { data: conv } = await ctx.supabase + .from('conversations') + .select('id') + .eq('id', id) + .eq('account_id', ctx.accountId) + .maybeSingle(); + if (!conv) return fail('not_found', 'Conversation not found', 404); + + let query = ctx.supabase + .from('messages') + .select('*') + .eq('conversation_id', id) + .order('created_at', { ascending: false }) + .order('id', { ascending: false }) + .limit(limit + 1); + + const kf = keysetFilter(cursor); + if (kf) query = query.or(kf); + + const { data, error } = await query; + if (error) { + console.error('[api/v1/messages] list error:', error); + return fail('internal', 'Failed to list messages', 500); + } + + const { items, nextCursor } = buildPage( + (data ?? []) as Array<{ created_at: string; id: string }>, + limit + ); + return okList( + items.map((m) => serializeMessage(m as unknown as Message)), + nextCursor + ); + } catch (err) { + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/conversations/[id]/route.ts b/src/app/api/v1/conversations/[id]/route.ts new file mode 100644 index 0000000..cf50746 --- /dev/null +++ b/src/app/api/v1/conversations/[id]/route.ts @@ -0,0 +1,40 @@ +// ============================================================ +// GET /api/v1/conversations/{id} — read one conversation +// (scope: conversations:read). Account-scoped: a foreign id → 404. +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { ok, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { + CONVERSATION_SELECT, + normalizeConversation, +} from '@/lib/inbox/conversations'; +import { serializeConversation } from '@/lib/api/v1/conversations'; +import type { Conversation } from '@/types'; + +export async function GET( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + try { + const ctx = await requireApiKey(request, 'conversations:read'); + const { id } = await params; + + const { data, error } = await ctx.supabase + .from('conversations') + .select(CONVERSATION_SELECT) + .eq('id', id) + .eq('account_id', ctx.accountId) + .maybeSingle(); + + if (error) { + console.error('[api/v1/conversations] read error:', error); + return fail('internal', 'Failed to read conversation', 500); + } + if (!data) return fail('not_found', 'Conversation not found', 404); + + return ok(serializeConversation(normalizeConversation(data as Conversation))); + } catch (err) { + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/conversations/route.ts b/src/app/api/v1/conversations/route.ts new file mode 100644 index 0000000..56146ca --- /dev/null +++ b/src/app/api/v1/conversations/route.ts @@ -0,0 +1,66 @@ +// ============================================================ +// GET /api/v1/conversations — list conversations (scope: conversations:read) +// +// Keyset-paginated (newest first). Filters: `?status=` (open/pending/ +// closed) and `?contact_id=`. Each conversation embeds its contact + +// tags via the shared CONVERSATION_SELECT. +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { okList, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { + parseListParams, + keysetFilter, + buildPage, +} from '@/lib/api/v1/pagination'; +import { + CONVERSATION_SELECT, + normalizeConversation, +} from '@/lib/inbox/conversations'; +import { serializeConversation } from '@/lib/api/v1/conversations'; +import type { Conversation } from '@/types'; + +export async function GET(request: Request) { + try { + const ctx = await requireApiKey(request, 'conversations:read'); + const { limit, cursor } = parseListParams(request); + const url = new URL(request.url); + const status = url.searchParams.get('status'); + const contactId = url.searchParams.get('contact_id'); + + let query = ctx.supabase + .from('conversations') + .select(CONVERSATION_SELECT) + .eq('account_id', ctx.accountId); + + if (status) query = query.eq('status', status); + if (contactId) query = query.eq('contact_id', contactId); + + query = query + .order('created_at', { ascending: false }) + .order('id', { ascending: false }) + .limit(limit + 1); + + const kf = keysetFilter(cursor); + if (kf) query = query.or(kf); + + const { data, error } = await query; + if (error) { + console.error('[api/v1/conversations] list error:', error); + return fail('internal', 'Failed to list conversations', 500); + } + + const { items, nextCursor } = buildPage( + (data ?? []) as Array<{ created_at: string; id: string }>, + limit + ); + return okList( + items.map((r) => + serializeConversation(normalizeConversation(r as Conversation)) + ), + nextCursor + ); + } catch (err) { + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/v1/messages/route.ts b/src/app/api/v1/messages/route.ts new file mode 100644 index 0000000..f10937c --- /dev/null +++ b/src/app/api/v1/messages/route.ts @@ -0,0 +1,136 @@ +// ============================================================ +// POST /api/v1/messages — send a WhatsApp message via the public API. +// +// The headline public endpoint (issue #245). Unlike the dashboard's +// `/api/whatsapp/send` (which takes an internal `conversation_id`), +// this takes a phone number — what an external automation actually +// has — resolves-or-creates the contact + conversation, then runs the +// same shared send core. +// +// Auth: API key with the `messages:send` scope. Account context (and +// the service-role client) come from `requireApiKey`. +// +// Body: +// { +// "to": "+14155550123", // required, E.164 +// "type": "text", // text|template|image|video|document|audio (default: text) +// "text": "Hello!", // text body, or media caption +// "media_url": "https://…/file.pdf", // required for image/video/document/audio +// "filename": "invoice.pdf", // optional, document filename +// "template": { // required when type=template +// "name": "order_update", +// "language": "en_US", +// "params": ["A123"] | { "body": [...] } // array = positional body; object = structured +// }, +// "reply_to_message_id": "", // optional, must be in the same conversation +// "name": "Jane Doe" // optional, names a newly-created contact +// } +// +// Response (201): +// { "data": { "message_id", "whatsapp_message_id", "conversation_id", +// "contact_id", "contact_created" } } +// ============================================================ + +import { requireApiKey } from '@/lib/auth/api-context'; +import { ok, fail, toApiErrorResponse } from '@/lib/api/v1/respond'; +import { resolveConversationByPhone } from '@/lib/whatsapp/resolve-conversation'; +import { + sendMessageToConversation, + validateSendMessageParams, + SendMessageError, +} from '@/lib/whatsapp/send-message'; + +export async function POST(request: Request) { + try { + const ctx = await requireApiKey(request, 'messages:send'); + + const body = (await request.json().catch(() => null)) as Record< + string, + unknown + > | null; + if (!body || typeof body !== 'object') { + return fail('bad_request', 'Request body must be a JSON object', 400); + } + + const to = typeof body.to === 'string' ? body.to.trim() : ''; + if (!to) { + return fail('bad_request', "'to' is required", 400); + } + + const type = typeof body.type === 'string' ? body.type : 'text'; + + // Unpack the optional `template` object into the flat params the + // send core expects. `params` as an array → legacy positional body + // params; as an object → structured header/body/button params. + const template = + body.template && typeof body.template === 'object' + ? (body.template as Record) + : null; + const templateParams = Array.isArray(template?.params) + ? (template.params as unknown[]).filter( + (p): p is string => typeof p === 'string' + ) + : undefined; + const templateMessageParams = + template?.params && !Array.isArray(template.params) + ? template.params + : undefined; + + // Validate the message shape BEFORE resolveConversationByPhone + // finds-or-creates a contact + conversation, so a bad payload 400s + // without leaving an orphan contact/conversation behind. + validateSendMessageParams({ + messageType: type, + contentText: typeof body.text === 'string' ? body.text : null, + mediaUrl: typeof body.media_url === 'string' ? body.media_url : null, + templateName: typeof template?.name === 'string' ? template.name : null, + }); + + // Find-or-create the conversation for this phone, then send. Both + // steps share `SendMessageError`, so one catch maps the whole + // pipeline to the envelope. + const resolved = await resolveConversationByPhone( + ctx.supabase, + ctx.accountId, + to, + typeof body.name === 'string' ? body.name : null + ); + + const result = await sendMessageToConversation( + ctx.supabase, + ctx.accountId, + { + conversationId: resolved.conversationId, + messageType: type, + contentText: typeof body.text === 'string' ? body.text : null, + mediaUrl: typeof body.media_url === 'string' ? body.media_url : null, + filename: typeof body.filename === 'string' ? body.filename : null, + templateName: typeof template?.name === 'string' ? template.name : null, + templateLanguage: + typeof template?.language === 'string' ? template.language : null, + templateParams, + templateMessageParams, + replyToMessageId: + typeof body.reply_to_message_id === 'string' + ? body.reply_to_message_id + : null, + } + ); + + return ok( + { + message_id: result.messageId, + whatsapp_message_id: result.whatsappMessageId, + conversation_id: resolved.conversationId, + contact_id: resolved.contactId, + contact_created: resolved.contactCreated, + }, + 201 + ); + } catch (err) { + if (err instanceof SendMessageError) { + return fail(err.code, err.message, err.status); + } + return toApiErrorResponse(err); + } +} diff --git a/src/app/api/whatsapp/send/route.test.ts b/src/app/api/whatsapp/send/route.test.ts index dc0fb57..aa54717 100644 --- a/src/app/api/whatsapp/send/route.test.ts +++ b/src/app/api/whatsapp/send/route.test.ts @@ -14,6 +14,10 @@ const messageInserts: Array> = [] // Toggles for the per-test scenario. let existingConversation: Record | null = null let contactRow: Record | null = null +// A conversation created during the request becomes retrievable by id — +// the shared send core re-loads the conversation (with its contact) from +// just the id, so the mock must model insert-then-select-by-id. +let createdConversation: Record | null = null const CONTACT = { id: 'contact-1', @@ -35,7 +39,9 @@ function makeSupabaseMock() { case 'contacts': return { data: contactRow, error: null } case 'conversations': - return { data: existingConversation, error: null } + // Once created this request, a by-id reload returns it (with + // its contact); otherwise fall back to the canned existing row. + return { data: createdConversation ?? existingConversation, error: null } case 'whatsapp_config': return { data: { @@ -82,7 +88,15 @@ function makeSupabaseMock() { } b.insert = vi.fn((payload: Record) => { didInsert = true - if (table === 'conversations') conversationInserts.push(payload) + if (table === 'conversations') { + conversationInserts.push(payload) + createdConversation = { + id: 'conv-new', + account_id: 'acct-1', + contact_id: 'contact-1', + contact: CONTACT, + } + } if (table === 'messages') messageInserts.push(payload) return b }) @@ -163,6 +177,7 @@ describe('POST /api/whatsapp/send — contact_id template path', () => { conversationInserts.length = 0 messageInserts.length = 0 existingConversation = null + createdConversation = null contactRow = CONTACT supabaseMock = makeSupabaseMock() sendTemplateMessage.mockClear() diff --git a/src/app/api/whatsapp/send/route.ts b/src/app/api/whatsapp/send/route.ts index 681d9dc..cd9eac2 100644 --- a/src/app/api/whatsapp/send/route.ts +++ b/src/app/api/whatsapp/send/route.ts @@ -1,27 +1,25 @@ import { NextResponse } from 'next/server' import { createClient } from '@/lib/supabase/server' -import { - sendTextMessage, - sendTemplateMessage, - sendMediaMessage, - type MediaKind, -} from '@/lib/whatsapp/meta-api' -import { decrypt, encrypt, isLegacyFormat } from '@/lib/whatsapp/encryption' -import { supabaseAdmin } from '@/lib/flows/admin-client' -import { - sanitizePhoneForMeta, - isValidE164, - phoneVariants, - isRecipientNotAllowedError, -} from '@/lib/whatsapp/phone-utils' import { checkRateLimit, rateLimitResponse, RATE_LIMITS, } from '@/lib/rate-limit' -import type { MessageTemplate } from '@/types' -import { isMessageTemplate } from '@/lib/whatsapp/template-row-guard' +import { + sendMessageToConversation, + validateSendMessageParams, + SendMessageError, +} from '@/lib/whatsapp/send-message' +// The dashboard's outbound-send endpoint. It owns auth, per-user rate +// limiting, and the two ways the UI targets a thread — an existing +// `conversation_id` (inbox) or a `contact_id` (Contact detail → +// find-or-create the conversation). The actual Meta plumbing (validate +// → send → persist → pause flows) lives in the shared +// `sendMessageToConversation` core, which the public `/api/v1/messages` +// endpoint reuses. This route is a thin adapter: resolve the +// conversation, delegate, then map `SendMessageError` back onto the +// dashboard's internal `{ error }` shape. export async function POST(request: Request) { try { const supabase = await createClient() @@ -90,66 +88,33 @@ export async function POST(request: Request) { ) } - // Media kinds (image/video/document/audio) are sent to Meta via a - // public URL the composer already uploaded to the chat-media bucket. - const MEDIA_KINDS = ['image', 'video', 'document', 'audio'] as const - const isMediaKind = (MEDIA_KINDS as readonly string[]).includes(message_type) - - // Reject anything outside the known set up front rather than letting - // an unknown type fall through to the text path with empty content. - const VALID_MESSAGE_TYPES = ['text', 'template', ...MEDIA_KINDS] as const - if (!(VALID_MESSAGE_TYPES as readonly string[]).includes(message_type)) { - return NextResponse.json( - { error: `Unsupported message_type "${message_type}"` }, - { status: 400 } - ) - } - - if (message_type === 'text' && !content_text) { - return NextResponse.json( - { error: 'content_text is required for text messages' }, - { status: 400 } - ) - } - - if (message_type === 'template' && !template_name) { - return NextResponse.json( - { error: 'template_name is required for template messages' }, - { status: 400 } - ) - } - - if (isMediaKind && !media_url) { - return NextResponse.json( - { error: `media_url is required for ${message_type} messages` }, - { status: 400 } - ) - } - - // Meta caps media captions at 1024 chars; reject before the upload is - // wasted at the Meta call. (Audio carries no caption — see meta-api.) - if ( - isMediaKind && - message_type !== 'audio' && - typeof content_text === 'string' && - content_text.length > 1024 - ) { - return NextResponse.json( - { error: 'Caption exceeds the 1024-character limit' }, - { status: 400 } - ) + // Validate the message shape up front — before the contact_id path + // finds-or-creates a conversation — so an invalid payload 400s + // without leaving an orphan empty conversation behind. + try { + validateSendMessageParams({ + messageType: message_type, + contentText: content_text, + mediaUrl: media_url, + templateName: template_name, + }) + } catch (err) { + if (err instanceof SendMessageError) { + return NextResponse.json({ error: err.message }, { status: err.status }) + } + throw err } // Resolve the target conversation. With `conversation_id` we load the // existing thread; with `contact_id` we find-or-create one for the // contact so a business-initiated template send (Contact detail view) - // reuses this whole path — phone variants, send-builder, persistence. - let conversation: { id: string; contact?: { id: string; phone?: string } | null } | null = null + // reuses the shared send core below. + let conversationId: string | null = null if (conversationIdInput) { const { data, error: convError } = await supabase .from('conversations') - .select('*, contact:contacts(*)') + .select('id') .eq('id', conversationIdInput) .eq('account_id', accountId) .single() @@ -160,13 +125,13 @@ export async function POST(request: Request) { { status: 404 } ) } - conversation = data + conversationId = data.id } else { // contact_id path: verify the contact is in this account first so a // caller can't open a conversation against someone else's contact. const { data: contactRow, error: contactErr } = await supabase .from('contacts') - .select('*') + .select('id') .eq('id', contact_id) .eq('account_id', accountId) .maybeSingle() @@ -190,305 +155,48 @@ export async function POST(request: Request) { { status: 500 } ) } - // The embed may not round-trip on insert; pin the contact we verified. - conversation = { ...resolved, contact: resolved.contact ?? contactRow } + conversationId = resolved } - if (!conversation) { + if (!conversationId) { return NextResponse.json( { error: 'Conversation not found' }, { status: 404 } ) } - const conversation_id = conversation.id - - const contact = conversation.contact - if (!contact?.phone) { - return NextResponse.json( - { error: 'Contact phone number not found' }, - { status: 400 } - ) - } - - // Sanitize and validate phone - const sanitizedPhone = sanitizePhoneForMeta(contact.phone) - if (!isValidE164(sanitizedPhone)) { - return NextResponse.json( - { error: 'Invalid phone number format' }, - { status: 400 } - ) - } - - // Fetch and decrypt WhatsApp config - const { data: config, error: configError } = await supabase - .from('whatsapp_config') - .select('*') - .eq('account_id', accountId) - .single() - - if (configError || !config) { - return NextResponse.json( - { error: 'WhatsApp not configured. Please set up your WhatsApp integration first.' }, - { status: 400 } - ) - } - - const accessToken = decrypt(config.access_token) - - // Self-heal legacy CBC-encrypted tokens. Fire-and-forget: we - // return from the send without waiting, so a failed upgrade just - // means the next send tries again. The upgrade is idempotent — - // concurrent sends both produce valid GCM ciphertexts of the same - // plaintext, last write wins. - if (isLegacyFormat(config.access_token)) { - void supabase - .from('whatsapp_config') - .update({ access_token: encrypt(accessToken) }) - .eq('id', config.id) - .then(({ error }) => { - if (error) { - console.warn( - '[whatsapp/send] access_token GCM upgrade failed:', - error.message, - ) - } - }) - } - - // Resolve the reply target (if any) to its Meta message_id, which is - // what `context.message_id` on the outgoing Meta payload needs. The - // parent must belong to this same conversation — otherwise a caller - // could quote messages they can't see by guessing UUIDs. - let contextMessageId: string | undefined - if (reply_to_message_id) { - const { data: parent, error: parentError } = await supabase - .from('messages') - .select('message_id, conversation_id') - .eq('id', reply_to_message_id) - .eq('conversation_id', conversation_id) - .maybeSingle() - - if (parentError || !parent) { - return NextResponse.json( - { error: 'reply_to_message_id not found in this conversation' }, - { status: 400 } - ) - } - if (!parent.message_id) { - // Parent never reached Meta (still in 'sending' or 'failed') — we - // can't quote it on WhatsApp. Send without context rather than - // dropping the message entirely. - console.warn( - '[whatsapp/send] reply target has no Meta message_id; sending without context' - ) - } else { - contextMessageId = parent.message_id - } - } - - // Send via Meta API — retry with phone-number variants if Meta rejects - // with "recipient not in allowed list" (common in sandbox / when a - // number was registered with/without a trunk 0). If an alternate - // format succeeds, we persist it back to the contact row so the - // next send goes through on the first attempt. - let waMessageId = '' - let workingPhone = sanitizedPhone - - // For template sends, load the row so sendTemplateMessage can - // build header + button components from the template definition. - // Match on (user_id, name, language) — same triple the unique - // index enforces — so multi-language templates work correctly. - // Missing template falls through with `templateRow = null` and - // the legacy body-only path runs. - // Load the template row so sendTemplateMessage can build header - // + button components from the definition. isMessageTemplate - // guards against a malformed row (e.g. from a partial sync) - // crashing the send-builder later in the stack. - let templateRow: MessageTemplate | null = null - if (message_type === 'template' && template_name) { - const { data } = await supabase - .from('message_templates') - .select('*') - .eq('account_id', accountId) - .eq('name', template_name) - .eq('language', template_language || 'en_US') - .maybeSingle() - if (data && !isMessageTemplate(data)) { - return NextResponse.json( - { - error: - 'Template row is malformed locally — run "Sync from Meta" in Settings to repair it.', - }, - { status: 500 }, - ) - } - templateRow = data ?? null - } - - const attempt = async (phone: string): Promise => { - if (message_type === 'template') { - const result = await sendTemplateMessage({ - phoneNumberId: config.phone_number_id, - accessToken, - to: phone, - templateName: template_name, - language: template_language || 'en_US', - template: templateRow ?? undefined, - messageParams: template_message_params ?? undefined, - // Legacy body-only fallback — only consulted when - // messageParams.body isn't set. - params: template_params || [], - contextMessageId, - }) - return result.messageId - } - if (isMediaKind) { - // content_text doubles as the caption (ignored for audio inside - // sendMediaMessage). filename surfaces in the recipient's chat - // for documents only. - const result = await sendMediaMessage({ - phoneNumberId: config.phone_number_id, - accessToken, - to: phone, - kind: message_type as MediaKind, - link: media_url, - caption: content_text || undefined, - filename: filename || undefined, - contextMessageId, - }) - return result.messageId - } - const result = await sendTextMessage({ - phoneNumberId: config.phone_number_id, - accessToken, - to: phone, - text: content_text, - contextMessageId, - }) - return result.messageId - } - + // Delegate to the shared send core (validates, sends to Meta with + // phone-variant retry, persists, pauses active flow runs). Its + // `SendMessageError` carries a machine code + HTTP status; the + // dashboard maps it to the internal `{ error }` shape. try { - const variants = phoneVariants(sanitizedPhone) - let lastError: unknown = null - - for (const variant of variants) { - try { - waMessageId = await attempt(variant) - workingPhone = variant - lastError = null - break - } catch (err) { - const message = err instanceof Error ? err.message : String(err) - // Only retry when the failure is specifically that the - // recipient isn't in Meta's allowed list. Any other error - // (bad token, invalid template, etc.) bubbles up immediately. - if (!isRecipientNotAllowedError(message)) { - throw err - } - lastError = err - console.warn(`[whatsapp/send] variant "${variant}" rejected by Meta, trying next…`) - } - } - - if (lastError) throw lastError - } catch (err) { - const message = err instanceof Error ? err.message : 'Unknown Meta API error' - console.error('Meta API send failed for all variants:', message) - return NextResponse.json( - { error: `Meta API error: ${message}` }, - { status: 502 } - ) - } - - // If a non-original variant succeeded, update the contact so future - // sends go straight through. sanitizePhoneForMeta on workingPhone - // will yield workingPhone itself, so re-storing preserves it. - if (workingPhone !== sanitizedPhone) { - console.log( - `[whatsapp/send] Auto-corrected contact phone: ${sanitizedPhone} → ${workingPhone}` - ) - await supabase - .from('contacts') - .update({ phone: workingPhone }) - .eq('id', contact.id) - } - - // Insert message into DB — field names MUST match the messages schema - // (see supabase/migrations/001_initial_schema.sql): - // conversation_id, sender_type, content_type, content_text, - // media_url, template_name, message_id, status, created_at - const { data: messageRecord, error: msgError } = await supabase - .from('messages') - .insert({ - conversation_id, - sender_type: 'agent', - content_type: message_type, - content_text: content_text || null, - media_url: media_url || null, - template_name: template_name || null, - message_id: waMessageId, - status: 'sent', - reply_to_message_id: reply_to_message_id || null, + const result = await sendMessageToConversation(supabase, accountId, { + conversationId, + messageType: message_type, + contentText: content_text, + mediaUrl: media_url, + filename, + templateName: template_name, + templateLanguage: template_language, + templateParams: template_params, + templateMessageParams: template_message_params, + replyToMessageId: reply_to_message_id, }) - .select() - .single() - if (msgError) { - console.error('Error inserting sent message:', msgError) - return NextResponse.json( - { error: `Message sent to Meta but failed to save to DB: ${msgError.message}` }, - { status: 500 } - ) - } - - // Update conversation - await supabase - .from('conversations') - .update({ - last_message_text: content_text || `[${message_type}]`, - last_message_at: new Date().toISOString(), - updated_at: new Date().toISOString(), + return NextResponse.json({ + success: true, + message_id: result.messageId, + whatsapp_message_id: result.whatsappMessageId, }) - .eq('id', conversation_id) - - // Pause any active Flow run for this contact — the agent stepping - // in is the strongest "yield, human is here" signal. See PR #2 - // plan for why we pause (not end): preserves diagnostic state + - // lets the agent or the 24h timeout sweep cleanly resolve the - // run later. For accounts with no active runs the UPDATE matches - // zero rows — cheap and harmless. - try { - const { error: pauseErr } = await supabaseAdmin() - .from('flow_runs') - .update({ - status: 'paused_by_agent', - ended_at: new Date().toISOString(), - end_reason: 'agent_replied', - }) - .eq('account_id', accountId) - .eq('contact_id', contact.id) - .eq('status', 'active') - if (pauseErr) { - // Best-effort — log + continue. The agent's message already - // landed at Meta; don't fail the response over a bookkeeping - // miss. Worst case: a stale active run gets caught by the - // stale-run cron sweep within 24h. - console.error('[flows] pause-on-agent-send failed:', pauseErr.message) - } } catch (err) { - console.error( - '[flows] pause-on-agent-send threw:', - err instanceof Error ? err.message : err, - ) + if (err instanceof SendMessageError) { + return NextResponse.json( + { error: err.message }, + { status: err.status } + ) + } + throw err } - - return NextResponse.json({ - success: true, - message_id: messageRecord.id, - whatsapp_message_id: waMessageId, - }) } catch (error) { console.error('Error in WhatsApp send POST:', error) return NextResponse.json( @@ -501,8 +209,8 @@ export async function POST(request: Request) { type SendSupabase = Awaited> /** - * Return the contact's conversation in this account, creating one if it - * doesn't exist yet. Mirrors the webhook's find-or-create so an + * Return the contact's conversation id in this account, creating one if + * it doesn't exist yet. Mirrors the webhook's find-or-create so an * inbound-then-outbound (or outbound-first) sequence converges on a single * thread per contact. Runs under the caller's RLS — the conversations_insert * policy requires account agent membership, which the caller already is. @@ -512,15 +220,15 @@ async function findOrCreateConversation( accountId: string, userId: string, contactId: string, -) { +): Promise { const { data: existing } = await supabase .from('conversations') - .select('*, contact:contacts(*)') + .select('id') .eq('account_id', accountId) .eq('contact_id', contactId) .maybeSingle() - if (existing) return existing + if (existing) return existing.id const { data: created, error } = await supabase .from('conversations') @@ -529,7 +237,7 @@ async function findOrCreateConversation( user_id: userId, contact_id: contactId, }) - .select('*, contact:contacts(*)') + .select('id') .single() if (error) { @@ -537,5 +245,5 @@ async function findOrCreateConversation( return null } - return created + return created.id } diff --git a/src/lib/api/v1/contacts.test.ts b/src/lib/api/v1/contacts.test.ts new file mode 100644 index 0000000..7a6ea91 --- /dev/null +++ b/src/lib/api/v1/contacts.test.ts @@ -0,0 +1,65 @@ +import { describe, it, expect } from 'vitest'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { + serializeContact, + findOrCreateContact, + ContactError, +} from './contacts'; + +describe('serializeContact', () => { + it('flattens contact_tags(tags(*)) onto a tags array and nulls missing fields', () => { + const row = { + id: 'c1', + phone: '+14155550123', + name: 'Jane', + email: null, + company: 'Acme', + avatar_url: null, + created_at: '2026-01-01T00:00:00Z', + updated_at: '2026-01-02T00:00:00Z', + contact_tags: [ + { tags: { id: 't1', name: 'vip', color: '#fff' } }, + { tags: null }, // orphaned join — dropped + ], + }; + expect(serializeContact(row)).toEqual({ + id: 'c1', + phone: '+14155550123', + name: 'Jane', + email: null, + company: 'Acme', + avatar_url: null, + tags: [{ id: 't1', name: 'vip', color: '#fff' }], + created_at: '2026-01-01T00:00:00Z', + updated_at: '2026-01-02T00:00:00Z', + }); + }); + + it('tolerates a row with no contact_tags key', () => { + const row = { + id: 'c2', + phone: '+1', + name: null, + email: null, + company: null, + avatar_url: null, + created_at: 'a', + updated_at: 'b', + }; + expect(serializeContact(row).tags).toEqual([]); + }); +}); + +describe('findOrCreateContact', () => { + const noopDb = {} as SupabaseClient; + + it('rejects a non-E.164 phone with a 400 ContactError', async () => { + await expect( + findOrCreateContact(noopDb, 'acc', 'user', { phone: 'not-a-number' }) + ).rejects.toMatchObject({ status: 400 }); + await expect( + findOrCreateContact(noopDb, 'acc', 'user', { phone: 'not-a-number' }) + ).rejects.toBeInstanceOf(ContactError); + }); +}); diff --git a/src/lib/api/v1/contacts.ts b/src/lib/api/v1/contacts.ts new file mode 100644 index 0000000..a37d972 --- /dev/null +++ b/src/lib/api/v1/contacts.ts @@ -0,0 +1,223 @@ +// ============================================================ +// Shared contact logic for the public API (v1) contact endpoints. +// +// Kept out of the route files so `GET/POST /api/v1/contacts` and +// `GET/PATCH /api/v1/contacts/{id}` share one serializer, one +// find-or-create (built on the same `findExistingContact` dedupe the +// webhook and send path use), and one tag-sync routine. +// ============================================================ + +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { findExistingContact, isUniqueViolation } from '@/lib/contacts/dedupe'; +import { resolveImportTagIds } from '@/lib/contacts/resolve-import-tags'; +import { sanitizePhoneForMeta, isValidE164 } from '@/lib/whatsapp/phone-utils'; + +/** Row select that embeds the contact's tags for serialization. */ +export const CONTACT_SELECT = '*, contact_tags(tags(*))'; + +export interface ApiContact { + id: string; + phone: string; + name: string | null; + email: string | null; + company: string | null; + avatar_url: string | null; + tags: { id: string; name: string; color: string }[]; + created_at: string; + updated_at: string; +} + +/** Thrown by the helpers below; routes map `.status`/`.message`. */ +export class ContactError extends Error { + readonly status: number; + constructor(message: string, status: number) { + super(message); + this.name = 'ContactError'; + this.status = status; + } +} + +type RawTagJoin = { tags: { id: string; name: string; color: string } | null }; + +/** Flatten a `CONTACT_SELECT` row into the public contact shape. */ +export function serializeContact(row: Record): ApiContact { + const joins = (row.contact_tags as RawTagJoin[] | undefined) ?? []; + return { + id: row.id as string, + phone: row.phone as string, + name: (row.name as string | null) ?? null, + email: (row.email as string | null) ?? null, + company: (row.company as string | null) ?? null, + avatar_url: (row.avatar_url as string | null) ?? null, + tags: joins + .map((j) => j.tags) + .filter((t): t is NonNullable => t != null) + .map((t) => ({ id: t.id, name: t.name, color: t.color })), + created_at: row.created_at as string, + updated_at: row.updated_at as string, + }; +} + +/** + * Resolve the audit `user_id` for API-created rows — the SINGLE source + * of truth used by every public-API write (contacts, messages, + * broadcasts, resolve-conversation), so the same key's writes are + * always attributed to the same human. API callers have no logged-in + * user, so — like the inbound webhook — we attribute writes to the + * **WhatsApp config owner** (the webhook's own convention). Contacts + * can be created before WhatsApp is connected, so we fall back to the + * account owner when there's no config yet. + */ +export async function resolveAuditUserId( + db: SupabaseClient, + accountId: string +): Promise { + const { data: config } = await db + .from('whatsapp_config') + .select('user_id') + .eq('account_id', accountId) + .maybeSingle(); + const configOwner = config?.user_id as string | undefined; + if (configOwner) return configOwner; + + const { data: account } = await db + .from('accounts') + .select('owner_user_id') + .eq('id', accountId) + .maybeSingle(); + const owner = account?.owner_user_id as string | undefined; + if (!owner) { + throw new ContactError('Account owner could not be resolved', 500); + } + return owner; +} + +export interface ContactInput { + phone: string; + name?: string | null; + email?: string | null; + company?: string | null; +} + +/** + * Find (by fuzzy phone match) or create a contact in `accountId`. + * Returns the contact id and whether it was created. Reuses the shared + * `findExistingContact` dedupe + unique-violation race backstop so an + * API-created contact is indistinguishable from a webhook-created one. + */ +export async function findOrCreateContact( + db: SupabaseClient, + accountId: string, + auditUserId: string, + input: ContactInput +): Promise<{ id: string; created: boolean }> { + const sanitized = sanitizePhoneForMeta(input.phone); + if (!isValidE164(sanitized)) { + throw new ContactError( + "'phone' must be a valid phone number in E.164 format (e.g. +14155550123)", + 400 + ); + } + + const existing = await findExistingContact(db, accountId, sanitized); + if (existing) return { id: existing.id, created: false }; + + const { data: created, error } = await db + .from('contacts') + .insert({ + account_id: accountId, + user_id: auditUserId, + phone: sanitized, + name: input.name ?? sanitized, + email: input.email ?? null, + company: input.company ?? null, + }) + .select('id') + .single(); + + if (error || !created) { + // Lost a race against a concurrent create — the unique index + // rejected the duplicate. Re-resolve to the winner. + if (isUniqueViolation(error)) { + const raced = await findExistingContact(db, accountId, sanitized); + if (raced) return { id: raced.id, created: false }; + } + console.error('[api/v1/contacts] create error:', error); + throw new ContactError('Failed to create contact', 500); + } + + return { id: created.id, created: true }; +} + +/** + * Replace a contact's tags to exactly match `tagNames` (case- + * insensitive; missing tags are created). A no-op when `tagNames` is + * undefined — pass `[]` to clear all tags. Reuses `resolveImportTagIds` + * so API and CSV-import tag handling stay consistent. + */ +export async function setContactTags( + db: SupabaseClient, + accountId: string, + auditUserId: string, + contactId: string, + tagNames: string[] +): Promise { + const { tagIdByKey } = await resolveImportTagIds(db, { + accountId, + userId: auditUserId, + tagNames, + canCreateTags: true, + }); + const desired = new Set(tagIdByKey.values()); + + // Diff against the current joins rather than delete-all-then-insert: + // a diff only touches tags that actually change, so a mid-operation + // failure can never wipe tags that were meant to stay. Every write + // is error-checked and surfaced as a ContactError (→ 500) instead of + // being swallowed behind a misleading 200. + const { data: current, error: readErr } = await db + .from('contact_tags') + .select('tag_id') + .eq('contact_id', contactId); + if (readErr) { + throw new ContactError('Failed to read contact tags', 500); + } + const existing = new Set( + (current ?? []).map((r) => r.tag_id as string) + ); + + const toAdd = [...desired].filter((id) => !existing.has(id)); + const toRemove = [...existing].filter((id) => !desired.has(id)); + + if (toRemove.length > 0) { + const { error } = await db + .from('contact_tags') + .delete() + .eq('contact_id', contactId) + .in('tag_id', toRemove); + if (error) throw new ContactError('Failed to update contact tags', 500); + } + if (toAdd.length > 0) { + const { error } = await db + .from('contact_tags') + .insert(toAdd.map((tag_id) => ({ contact_id: contactId, tag_id }))); + if (error) throw new ContactError('Failed to update contact tags', 500); + } +} + +/** Fetch + serialize a single contact scoped to the account, or null. */ +export async function getContactById( + db: SupabaseClient, + accountId: string, + contactId: string +): Promise { + const { data, error } = await db + .from('contacts') + .select(CONTACT_SELECT) + .eq('id', contactId) + .eq('account_id', accountId) + .maybeSingle(); + if (error || !data) return null; + return serializeContact(data as Record); +} diff --git a/src/lib/api/v1/conversations.test.ts b/src/lib/api/v1/conversations.test.ts new file mode 100644 index 0000000..8136dca --- /dev/null +++ b/src/lib/api/v1/conversations.test.ts @@ -0,0 +1,54 @@ +import { describe, it, expect } from 'vitest'; +import type { Conversation, Message } from '@/types'; +import { serializeConversation, serializeMessage } from './conversations'; + +describe('serializeConversation', () => { + it('projects public fields + nested contact/tags and drops internals', () => { + const conv = { + id: 'conv1', + user_id: 'internal-user', + account_id: 'internal-acct', + contact_id: 'c1', + status: 'open', + last_message_text: 'hi', + last_message_at: '2026-01-01T00:00:00Z', + unread_count: 2, + created_at: '2026-01-01T00:00:00Z', + updated_at: '2026-01-01T00:00:00Z', + contact: { + id: 'c1', + phone: '+1', + name: 'Jane', + tags: [{ id: 't1', name: 'vip', color: '#fff' }], + }, + } as unknown as Conversation; + + const out = serializeConversation(conv); + expect(out).not.toHaveProperty('user_id'); + expect(out).not.toHaveProperty('account_id'); + expect(out.contact?.tags).toEqual([{ id: 't1', name: 'vip', color: '#fff' }]); + expect(out.unread_count).toBe(2); + }); +}); + +describe('serializeMessage', () => { + it('maps message_id → whatsapp_message_id and derives direction', () => { + const inbound = { + id: 'm1', + conversation_id: 'conv1', + sender_type: 'customer', + content_type: 'text', + content_text: 'hello', + message_id: 'wamid.123', + status: 'delivered', + created_at: '2026-01-01T00:00:00Z', + } as unknown as Message; + const outMsg = serializeMessage(inbound); + expect(outMsg.direction).toBe('inbound'); + expect(outMsg.whatsapp_message_id).toBe('wamid.123'); + expect(outMsg).not.toHaveProperty('message_id'); + + const agent = { ...inbound, sender_type: 'agent' } as unknown as Message; + expect(serializeMessage(agent).direction).toBe('outbound'); + }); +}); diff --git a/src/lib/api/v1/conversations.ts b/src/lib/api/v1/conversations.ts new file mode 100644 index 0000000..f361051 --- /dev/null +++ b/src/lib/api/v1/conversations.ts @@ -0,0 +1,100 @@ +// ============================================================ +// Public API (v1) serializers for conversations + messages. +// +// The dashboard's `Conversation`/`Message` rows carry internal columns +// (account_id, user_id, sender_id) that shouldn't leak onto the public +// wire. These serializers project the stable public subset and rename +// the Meta id (`message_id` → `whatsapp_message_id`) to match the send +// endpoint's response vocabulary. +// ============================================================ + +import type { Conversation, Message } from '@/types'; + +export interface ApiConversation { + id: string; + contact_id: string; + status: string; + assigned_agent_id: string | null; + last_message_text: string | null; + last_message_at: string | null; + unread_count: number; + created_at: string; + updated_at: string; + contact: { + id: string; + phone: string; + name: string | null; + email: string | null; + company: string | null; + tags: { id: string; name: string; color: string }[]; + } | null; +} + +export interface ApiMessage { + id: string; + conversation_id: string; + direction: 'inbound' | 'outbound'; + sender_type: string; + content_type: string; + content_text: string | null; + media_url: string | null; + template_name: string | null; + whatsapp_message_id: string | null; + status: string; + reply_to_message_id: string | null; + interactive_reply_id: string | null; + created_at: string; +} + +/** + * Project a normalized `Conversation` (from `normalizeConversation`, + * which has already flattened `contact.tags`) into the public shape. + */ +export function serializeConversation(conv: Conversation): ApiConversation { + const c = conv.contact; + return { + id: conv.id, + contact_id: conv.contact_id, + status: conv.status, + assigned_agent_id: conv.assigned_agent_id ?? null, + last_message_text: conv.last_message_text ?? null, + last_message_at: conv.last_message_at ?? null, + unread_count: conv.unread_count ?? 0, + created_at: conv.created_at, + updated_at: conv.updated_at, + contact: c + ? { + id: c.id, + phone: c.phone, + name: c.name ?? null, + email: c.email ?? null, + company: c.company ?? null, + tags: (c.tags ?? []).map((t) => ({ + id: t.id, + name: t.name, + color: t.color, + })), + } + : null, + }; +} + +/** Project a `messages` row into the public shape. */ +export function serializeMessage(m: Message): ApiMessage { + return { + id: m.id, + conversation_id: m.conversation_id, + // `customer` = inbound (from the contact); anything else is outbound. + direction: m.sender_type === 'customer' ? 'inbound' : 'outbound', + sender_type: m.sender_type, + content_type: m.content_type, + content_text: m.content_text ?? null, + media_url: m.media_url ?? null, + template_name: m.template_name ?? null, + whatsapp_message_id: m.message_id ?? null, + status: m.status, + reply_to_message_id: m.reply_to_message_id ?? null, + interactive_reply_id: m.interactive_reply_id ?? null, + created_at: m.created_at, + }; +} diff --git a/src/lib/api/v1/pagination.test.ts b/src/lib/api/v1/pagination.test.ts new file mode 100644 index 0000000..118eaf0 --- /dev/null +++ b/src/lib/api/v1/pagination.test.ts @@ -0,0 +1,112 @@ +import { describe, it, expect } from 'vitest'; +import { + parseListParams, + encodeCursor, + decodeCursor, + keysetFilter, + buildPage, + DEFAULT_LIMIT, + MAX_LIMIT, +} from './pagination'; + +const req = (qs: string) => new Request(`https://x.test/api/v1/contacts${qs}`); + +describe('parseListParams', () => { + it('defaults limit and cursor', () => { + expect(parseListParams(req(''))).toEqual({ + limit: DEFAULT_LIMIT, + cursor: null, + }); + }); + + it('clamps limit to MAX_LIMIT and floors it', () => { + expect(parseListParams(req('?limit=9999')).limit).toBe(MAX_LIMIT); + expect(parseListParams(req('?limit=10.9')).limit).toBe(10); + }); + + it('falls back to default on non-positive / NaN limit', () => { + expect(parseListParams(req('?limit=0')).limit).toBe(DEFAULT_LIMIT); + expect(parseListParams(req('?limit=-5')).limit).toBe(DEFAULT_LIMIT); + expect(parseListParams(req('?limit=abc')).limit).toBe(DEFAULT_LIMIT); + }); + + it('decodes a valid cursor and ignores a malformed one', () => { + const c = encodeCursor({ + created_at: '2026-01-01T00:00:00Z', + id: '11111111-1111-4111-8111-111111111111', + }); + expect(parseListParams(req(`?cursor=${c}`)).cursor).toEqual({ + createdAt: '2026-01-01T00:00:00Z', + id: '11111111-1111-4111-8111-111111111111', + }); + expect(parseListParams(req('?cursor=@@notbase64@@')).cursor).toBeNull(); + }); +}); + +describe('encode/decodeCursor round-trip', () => { + it('round-trips a real (ISO timestamp, UUID) cursor', () => { + const row = { + created_at: '2026-06-30T12:00:00.123Z', + id: 'abcdef01-2345-4678-8abc-def012345678', + }; + expect(decodeCursor(encodeCursor(row))).toEqual({ + createdAt: '2026-06-30T12:00:00.123Z', + id: 'abcdef01-2345-4678-8abc-def012345678', + }); + }); + + it('returns null for empty / separator-less input', () => { + expect(decodeCursor(null)).toBeNull(); + expect(decodeCursor('')).toBeNull(); + expect(decodeCursor(Buffer.from('nosep').toString('base64url'))).toBeNull(); + }); + + it('rejects a crafted cursor whose id is not a UUID (filter-injection guard)', () => { + // A hand-built cursor trying to smuggle PostgREST filter syntax + // through keysetFilter must be refused, not decoded. + const evil = Buffer.from( + '2026-01-01T00:00:00Z|x),or(account_id.neq.0', + 'utf8' + ).toString('base64url'); + expect(decodeCursor(evil)).toBeNull(); + }); + + it('rejects a cursor whose timestamp is not parseable', () => { + const bad = Buffer.from( + 'not-a-date|11111111-1111-4111-8111-111111111111', + 'utf8' + ).toString('base64url'); + expect(decodeCursor(bad)).toBeNull(); + }); +}); + +describe('keysetFilter', () => { + it('is null on the first page', () => { + expect(keysetFilter(null)).toBeNull(); + }); + + it('walks strictly past the cursor row (older, or same-ts smaller id)', () => { + expect(keysetFilter({ createdAt: '2026-01-01T00:00:00Z', id: 'x' })).toBe( + 'created_at.lt.2026-01-01T00:00:00Z,and(created_at.eq.2026-01-01T00:00:00Z,id.lt.x)' + ); + }); +}); + +describe('buildPage', () => { + const rows = Array.from({ length: 6 }, (_, i) => ({ + created_at: `2026-01-0${i + 1}T00:00:00Z`, + id: `id${i + 1}`, + })); + + it('returns all rows and null cursor when not over-fetched', () => { + const page = buildPage(rows.slice(0, 3), 5); + expect(page.items).toHaveLength(3); + expect(page.nextCursor).toBeNull(); + }); + + it('trims to limit and emits a cursor for the last kept row', () => { + const page = buildPage(rows, 5); + expect(page.items).toHaveLength(5); + expect(page.nextCursor).toBe(encodeCursor(rows[4])); + }); +}); diff --git a/src/lib/api/v1/pagination.ts b/src/lib/api/v1/pagination.ts new file mode 100644 index 0000000..65ded12 --- /dev/null +++ b/src/lib/api/v1/pagination.ts @@ -0,0 +1,123 @@ +// ============================================================ +// Cursor pagination for public API (v1) list endpoints. +// +// Every `/api/v1` list route (contacts, conversations, messages, +// broadcasts) pages the same way so integrators write one loop: +// +// GET /api/v1/contacts?limit=50 +// → { "data": [...], "meta": { "next_cursor": "…" } } +// GET /api/v1/contacts?limit=50&cursor=… // next page +// → { "data": [...], "meta": { "next_cursor": null } } // last page +// +// Cursors are **keyset** (not offset): rows are ordered by +// `(created_at, id)` descending and the cursor encodes the last row's +// `(created_at, id)`. This is stable under concurrent inserts (an +// offset would skip/repeat rows when new data lands mid-scan) and +// stays fast at any depth. The cursor is an opaque base64 string — +// clients pass it back verbatim and never parse it. +// ============================================================ + +export const DEFAULT_LIMIT = 50; +export const MAX_LIMIT = 100; + +export interface Cursor { + createdAt: string; + id: string; +} + +export interface ListParams { + /** Clamped to [1, MAX_LIMIT]. */ + limit: number; + /** Decoded cursor, or null on the first page. */ + cursor: Cursor | null; +} + +/** + * Parse `?limit` and `?cursor` off a request URL. `limit` is clamped + * to [1, MAX_LIMIT] (default {@link DEFAULT_LIMIT}); a malformed or + * unparseable `cursor` is treated as absent (first page) rather than + * erroring — the worst case is the client re-reads from the top. + */ +export function parseListParams(request: Request): ListParams { + const url = new URL(request.url); + + const rawLimit = Number(url.searchParams.get('limit')); + const limit = + Number.isFinite(rawLimit) && rawLimit > 0 + ? Math.min(Math.floor(rawLimit), MAX_LIMIT) + : DEFAULT_LIMIT; + + return { limit, cursor: decodeCursor(url.searchParams.get('cursor')) }; +} + +/** Encode a row's `(created_at, id)` into an opaque cursor string. */ +export function encodeCursor(row: { created_at: string; id: string }): string { + return Buffer.from(`${row.created_at}|${row.id}`, 'utf8').toString( + 'base64url' + ); +} + +// A cursor is only ever minted by `encodeCursor` from a real row's +// `created_at` (ISO timestamp) + `id` (UUID). We re-validate both on +// decode so a hand-crafted cursor can't smuggle PostgREST filter +// syntax into `keysetFilter`'s `.or()` string (the values are +// interpolated raw). Anything that doesn't look server-issued is +// treated as "no cursor" — the documented tolerance is to restart +// from the first page, never to run an attacker-shaped query. +const UUID_RE = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +/** Decode a cursor string, or null if missing/malformed/untrusted. */ +export function decodeCursor(value: string | null): Cursor | null { + if (!value) return null; + try { + const decoded = Buffer.from(value, 'base64url').toString('utf8'); + const sep = decoded.indexOf('|'); + if (sep === -1) return null; + const createdAt = decoded.slice(0, sep); + const id = decoded.slice(sep + 1); + // Reject anything that isn't a plausible server-issued cursor: an + // ISO-8601 timestamp and a UUID. This is what keeps the raw + // interpolation in `keysetFilter` safe. + if (!UUID_RE.test(id)) return null; + const ts = Date.parse(createdAt); + if (Number.isNaN(ts)) return null; + return { createdAt, id }; + } catch { + return null; + } +} + +/** + * PostgREST `.or()` expression that walks *past* the cursor row under + * a `(created_at desc, id desc)` ordering: strictly-older rows, plus + * same-timestamp rows with a smaller id (the tie-breaker). Returns + * null on the first page. Apply as: + * + * let q = db.from('contacts').select('*').eq('account_id', accountId) + * .order('created_at', { ascending: false }) + * .order('id', { ascending: false }) + * .limit(limit + 1) // fetch one extra to detect a next page + * const f = keysetFilter(cursor) + * if (f) q = q.or(f) + */ +export function keysetFilter(cursor: Cursor | null): string | null { + if (!cursor) return null; + return `created_at.lt.${cursor.createdAt},and(created_at.eq.${cursor.createdAt},id.lt.${cursor.id})`; +} + +/** + * Trim an over-fetched result set (query ran with `limit + 1`) down to + * `limit` and derive the `next_cursor`. When fewer than `limit + 1` + * rows came back, this is the last page and `nextCursor` is null. + */ +export function buildPage( + rows: T[], + limit: number +): { items: T[]; nextCursor: string | null } { + if (rows.length <= limit) { + return { items: rows, nextCursor: null }; + } + const items = rows.slice(0, limit); + return { items, nextCursor: encodeCursor(items[items.length - 1]) }; +} diff --git a/src/lib/api/v1/respond.ts b/src/lib/api/v1/respond.ts index 49776f2..ac19c2b 100644 --- a/src/lib/api/v1/respond.ts +++ b/src/lib/api/v1/respond.ts @@ -86,6 +86,33 @@ export function ok(data: T, status = 200): NextResponse { return NextResponse.json({ data }, { status }); } +/** + * List envelope: `{ data: [...], meta: { next_cursor } }`. The `meta` + * block is the pagination contract shared by every v1 list endpoint — + * `next_cursor` is an opaque string to pass back as `?cursor=`, or + * `null` on the last page. See `src/lib/api/v1/pagination.ts`. + */ +export function okList(items: T[], nextCursor: string | null): NextResponse { + return NextResponse.json({ data: items, meta: { next_cursor: nextCursor } }); +} + +/** + * Failure envelope from an explicit (code, message, status). Use for + * domain errors whose codes live outside `ApiErrorCode` (e.g. the + * send pipeline's `meta_error` / `whatsapp_not_configured`) — the + * wire `code` is a free string, so any machine-meaningful value is + * fine. `headers` is rarely needed; omit unless you have a + * `Retry-After`-style set. + */ +export function fail( + code: string, + message: string, + status: number, + headers?: Record +): NextResponse { + return NextResponse.json({ error: { code, message } }, { status, headers }); +} + /** * Map any thrown value to the failure envelope. `ApiError` keeps its * code/status/headers; anything else collapses to a generic 500 so we diff --git a/src/lib/whatsapp/broadcast-core.test.ts b/src/lib/whatsapp/broadcast-core.test.ts new file mode 100644 index 0000000..4b4e449 --- /dev/null +++ b/src/lib/whatsapp/broadcast-core.test.ts @@ -0,0 +1,36 @@ +import { describe, it, expect } from 'vitest'; +import type { SupabaseClient } from '@supabase/supabase-js'; +import { createBroadcast, BroadcastError } from './broadcast-core'; + +// These assertions all fire in the pure validation prologue, before +// any Supabase call — a bare stub is enough. +const db = {} as SupabaseClient; + +describe('createBroadcast validation', () => { + it('rejects a missing template_name', async () => { + await expect( + createBroadcast(db, 'acc', 'user', { + templateName: '', + recipients: [{ to: '+14155550123' }], + }) + ).rejects.toMatchObject({ code: 'bad_request', status: 400 }); + }); + + it('rejects an empty recipient list', async () => { + await expect( + createBroadcast(db, 'acc', 'user', { + templateName: 'promo', + recipients: [], + }) + ).rejects.toBeInstanceOf(BroadcastError); + }); + + it('rejects more than 1000 recipients', async () => { + const recipients = Array.from({ length: 1001 }, () => ({ + to: '+14155550123', + })); + await expect( + createBroadcast(db, 'acc', 'user', { templateName: 'promo', recipients }) + ).rejects.toMatchObject({ status: 400 }); + }); +}); diff --git a/src/lib/whatsapp/broadcast-core.ts b/src/lib/whatsapp/broadcast-core.ts new file mode 100644 index 0000000..672ce6a --- /dev/null +++ b/src/lib/whatsapp/broadcast-core.ts @@ -0,0 +1,327 @@ +// ============================================================ +// Public-API broadcast core. +// +// Splits a broadcast into two phases so the HTTP route can persist + +// acknowledge fast and fan out afterwards (in `after()`): +// +// createBroadcast() — validate, resolve contacts, insert the +// `broadcasts` row + `broadcast_recipients` +// rows (status 'pending'), return a plan. +// deliverBroadcast() — send each recipient's template via Meta +// (phone-variant retry), stamp each recipient +// row + the aggregate counts, finalize status. +// +// Recipient rows carry `whatsapp_message_id`, so the inbound webhook's +// status handler (which matches on that column) updates delivered/read +// for API broadcasts exactly as it does for dashboard ones. +// ============================================================ + +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { sendTemplateMessage } from '@/lib/whatsapp/meta-api'; +import { decrypt } from '@/lib/whatsapp/encryption'; +import { + sanitizePhoneForMeta, + isValidE164, + phoneVariants, + isRecipientNotAllowedError, +} from '@/lib/whatsapp/phone-utils'; +import { isMessageTemplate } from '@/lib/whatsapp/template-row-guard'; +import type { MessageTemplate } from '@/types'; +import { findOrCreateContact } from '@/lib/api/v1/contacts'; + +/** Thrown by createBroadcast on a caller-visible failure; route maps it. */ +export class BroadcastError extends Error { + readonly code: string; + readonly status: number; + constructor(code: string, message: string, status: number) { + super(message); + this.name = 'BroadcastError'; + this.code = code; + this.status = status; + } +} + +export interface BroadcastRecipientInput { + /** E.164 phone. */ + to: string; + /** Positional body params for the template ({{1}}, {{2}}…). */ + params?: string[]; +} + +export interface CreateBroadcastParams { + name?: string | null; + templateName: string; + templateLanguage?: string | null; + recipients: BroadcastRecipientInput[]; +} + +interface PlannedRecipient { + recipientRowId: string; + phone: string; + params: string[]; +} + +export interface BroadcastPlan { + broadcastId: string; + templateName: string; + templateLanguage: string; + phoneNumberId: string; + accessToken: string; + templateRow: MessageTemplate | null; + planned: PlannedRecipient[]; + /** Phones rejected up front (invalid E.164) — counted as failed. */ + rejected: number; +} + +const MAX_RECIPIENTS = 1000; + +/** + * Validate + persist a broadcast, resolving each recipient to a + * contact. Returns a plan for {@link deliverBroadcast}. Throws + * {@link BroadcastError} on bad input / missing config / a malformed + * template / a DB failure — nothing is sent in this phase. + */ +export async function createBroadcast( + db: SupabaseClient, + accountId: string, + auditUserId: string, + params: CreateBroadcastParams +): Promise { + const { name, templateName, recipients } = params; + const templateLanguage = params.templateLanguage || 'en_US'; + + if (!templateName) { + throw new BroadcastError('bad_request', "'template_name' is required", 400); + } + if (!Array.isArray(recipients) || recipients.length === 0) { + throw new BroadcastError( + 'bad_request', + "'recipients' must be a non-empty array of { to, params? }", + 400 + ); + } + if (recipients.length > MAX_RECIPIENTS) { + throw new BroadcastError( + 'bad_request', + `A broadcast is capped at ${MAX_RECIPIENTS} recipients per request; split larger sends`, + 400 + ); + } + + // Config (fail fast + provides the audit trail owner already resolved + // by the caller). Meta send needs phone_number_id + decrypted token. + const { data: config, error: configError } = await db + .from('whatsapp_config') + .select('*') + .eq('account_id', accountId) + .single(); + if (configError || !config) { + throw new BroadcastError( + 'whatsapp_not_configured', + 'WhatsApp not configured. Please set up your WhatsApp integration first.', + 400 + ); + } + const accessToken = decrypt(config.access_token); + + // Template row (once) for header/button components; guard a + // malformed local row rather than N identical opaque failures. + const { data: rawTemplateRow } = await db + .from('message_templates') + .select('*') + .eq('account_id', accountId) + .eq('name', templateName) + .eq('language', templateLanguage) + .maybeSingle(); + if (rawTemplateRow && !isMessageTemplate(rawTemplateRow)) { + throw new BroadcastError( + 'template_malformed', + 'Template row is malformed locally — run "Sync from Meta" in Settings to repair it before broadcasting.', + 500 + ); + } + const templateRow = (rawTemplateRow as MessageTemplate | null) ?? null; + + // Resolve each recipient to a contact. Invalid phones are dropped + // (counted as rejected) rather than aborting the whole broadcast. + const resolved: { contactId: string; phone: string; params: string[] }[] = []; + let rejected = 0; + for (const r of recipients) { + const sanitized = sanitizePhoneForMeta(typeof r.to === 'string' ? r.to : ''); + if (!isValidE164(sanitized)) { + rejected++; + continue; + } + const { id } = await findOrCreateContact(db, accountId, auditUserId, { + phone: sanitized, + }); + resolved.push({ + contactId: id, + phone: sanitized, + params: Array.isArray(r.params) + ? r.params.filter((p): p is string => typeof p === 'string') + : [], + }); + } + + // Collapse recipients that resolved to the SAME contact (the caller + // listed a phone twice, or two numbers fuzzy-matched to one contact). + // Keep the first occurrence so the contact is messaged once and its + // params aren't silently overwritten by a later duplicate — and so + // the row↔params pairing below (keyed by contact_id) is unambiguous. + const seenContact = new Set(); + const deduped = resolved.filter((r) => { + if (seenContact.has(r.contactId)) return false; + seenContact.add(r.contactId); + return true; + }); + + if (deduped.length === 0) { + throw new BroadcastError( + 'bad_request', + 'No recipients had a valid E.164 phone number', + 400 + ); + } + + // Persist the broadcast + its recipients. The count columns + // (sent/delivered/read/replied/failed) are owned by the DB aggregate + // trigger (migrations 003/005) and derived purely from + // broadcast_recipients rows — we deliberately do NOT seed them here + // (a manual value would be clobbered by the trigger on the first + // recipient change). `rejected` phones have no recipient row, so they + // are reported to the caller in the POST response, not in these + // persisted counts. + const { data: broadcast, error: bErr } = await db + .from('broadcasts') + .insert({ + account_id: accountId, + user_id: auditUserId, + name: name || `API broadcast (${templateName})`, + template_name: templateName, + template_language: templateLanguage, + status: 'sending', + total_recipients: deduped.length, + }) + .select('id') + .single(); + if (bErr || !broadcast) { + console.error('[broadcast-core] create broadcast error:', bErr); + throw new BroadcastError('internal', 'Failed to create broadcast', 500); + } + + const { data: recipientRows, error: rErr } = await db + .from('broadcast_recipients') + .insert( + deduped.map((r) => ({ + broadcast_id: broadcast.id, + contact_id: r.contactId, + status: 'pending' as const, + })) + ) + .select('id, contact_id'); + if (rErr || !recipientRows) { + console.error('[broadcast-core] create recipients error:', rErr); + throw new BroadcastError('internal', 'Failed to create broadcast', 500); + } + + // Pair each inserted recipient row back to its phone/params by + // contact_id — unambiguous now that duplicates are collapsed. + const byContact = new Map(deduped.map((r) => [r.contactId, r])); + const planned: PlannedRecipient[] = recipientRows.map((row) => { + const r = byContact.get(row.contact_id as string)!; + return { recipientRowId: row.id as string, phone: r.phone, params: r.params }; + }); + + return { + broadcastId: broadcast.id, + templateName, + templateLanguage, + phoneNumberId: config.phone_number_id, + accessToken, + templateRow, + planned, + rejected, + }; +} + +/** + * Fan out a {@link BroadcastPlan}: send each recipient's template + * (phone-variant retry) and stamp its `broadcast_recipients` row. + * Best-effort per recipient — one failure never aborts the rest. + * Designed to run inside `after()`. + * + * The per-status count columns on `broadcasts` are owned by the DB + * aggregate trigger (migrations 003/005): each recipient-row update + * below advances them automatically, and later Meta delivery/read + * webhooks keep advancing them. We therefore never write those columns + * here — only the terminal `status` — otherwise a manual value would + * race and clobber the trigger-maintained counts. + */ +export async function deliverBroadcast( + db: SupabaseClient, + plan: BroadcastPlan +): Promise { + let sentCount = 0; + + for (const recipient of plan.planned) { + const variants = phoneVariants(recipient.phone); + let sentMessageId: string | null = null; + let lastError: string | null = null; + + for (const variant of variants) { + try { + const result = await sendTemplateMessage({ + phoneNumberId: plan.phoneNumberId, + accessToken: plan.accessToken, + to: variant, + templateName: plan.templateName, + language: plan.templateLanguage, + template: plan.templateRow ?? undefined, + params: recipient.params, + }); + sentMessageId = result.messageId; + lastError = null; + break; + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error'; + lastError = message; + // Only a "recipient not allowed" error is worth another variant. + if (!isRecipientNotAllowedError(message)) break; + } + } + + if (sentMessageId) { + sentCount++; + await db + .from('broadcast_recipients') + .update({ + status: 'sent', + sent_at: new Date().toISOString(), + whatsapp_message_id: sentMessageId, + error_message: null, + }) + .eq('id', recipient.recipientRowId); + } else { + await db + .from('broadcast_recipients') + .update({ + status: 'failed', + error_message: lastError || 'Unknown error', + }) + .eq('id', recipient.recipientRowId); + } + } + + // Terminal status only — counts are trigger-owned (see the note + // above). If nothing sent, the broadcast failed outright; a partial + // send is still 'sent' (per-recipient failures show in failed_count). + await db + .from('broadcasts') + .update({ + status: sentCount > 0 ? 'sent' : 'failed', + updated_at: new Date().toISOString(), + }) + .eq('id', plan.broadcastId); +} diff --git a/src/lib/whatsapp/resolve-conversation.test.ts b/src/lib/whatsapp/resolve-conversation.test.ts new file mode 100644 index 0000000..e483d80 --- /dev/null +++ b/src/lib/whatsapp/resolve-conversation.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it } from 'vitest'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { resolveConversationByPhone } from './resolve-conversation'; +import { SendMessageError } from './send-message'; + +// ------------------------------------------------------------ +// Chainable Supabase stub, scripted per table. Terminal methods +// (like/maybeSingle/single) resolve to configured data; the builder +// itself is thenable so an awaited `update().eq()` resolves cleanly. +// ------------------------------------------------------------ +type ContactRow = { id: string; phone: string; name?: string | null }; + +interface Script { + config?: { user_id: string } | null; // whatsapp_config.maybeSingle + contactCandidates?: ContactRow[]; // contacts .like (same every call) + /** Per-call `.like` results — overrides contactCandidates. Lets a + * test simulate "miss, then hit" for the unique-race path. */ + contactCandidatesByCall?: ContactRow[][]; + insertedContactId?: string; // contacts insert -> single + insertContactError?: { code?: string } | null; + existingConversation?: { id: string } | null; // conversations select.maybeSingle + insertedConversationId?: string; // conversations insert -> single +} + +function makeDb(script: Script): SupabaseClient { + let table = ''; + let mode: 'select' | 'insert' | 'update' = 'select'; + let likeCalls = 0; + + const builder: Record = { + select: () => builder, + insert: () => { + mode = 'insert'; + return builder; + }, + update: () => { + mode = 'update'; + return builder; + }, + eq: () => builder, + like: () => { + const data = script.contactCandidatesByCall + ? (script.contactCandidatesByCall[likeCalls] ?? []) + : (script.contactCandidates ?? []); + likeCalls++; + return Promise.resolve({ data, error: null }); + }, + maybeSingle: () => { + if (table === 'whatsapp_config') + return Promise.resolve({ data: script.config ?? null, error: null }); + if (table === 'conversations' && mode === 'select') + return Promise.resolve({ + data: script.existingConversation ?? null, + error: null, + }); + return Promise.resolve({ data: null, error: null }); + }, + single: () => { + if (table === 'contacts' && mode === 'insert') { + if (script.insertContactError) + return Promise.resolve({ + data: null, + error: script.insertContactError, + }); + return Promise.resolve({ + data: { id: script.insertedContactId }, + error: null, + }); + } + if (table === 'conversations' && mode === 'insert') + return Promise.resolve({ + data: { id: script.insertedConversationId }, + error: null, + }); + return Promise.resolve({ data: null, error: null }); + }, + // Thenable: `await db.from().update().eq()` lands here. + then: (resolve: (v: { data: null; error: null }) => void) => + resolve({ data: null, error: null }), + }; + + return { + from: (t: string) => { + table = t; + mode = 'select'; + return builder; + }, + } as unknown as SupabaseClient; +} + +describe('resolveConversationByPhone', () => { + it('rejects an invalid phone before any DB call', async () => { + const db = { + from() { + throw new Error('should not query'); + }, + } as unknown as SupabaseClient; + await expect( + resolveConversationByPhone(db, 'acct', 'not-a-phone') + ).rejects.toBeInstanceOf(SendMessageError); + }); + + it('fails with whatsapp_not_configured when no config owner exists', async () => { + const db = makeDb({ config: null }); + await resolveConversationByPhone(db, 'acct', '+14155550123').catch( + (e: SendMessageError) => { + expect(e.code).toBe('whatsapp_not_configured'); + expect(e.status).toBe(400); + } + ); + await expect( + resolveConversationByPhone(db, 'acct', '+14155550123') + ).rejects.toBeInstanceOf(SendMessageError); + }); + + it('returns the existing contact + conversation without creating', async () => { + const db = makeDb({ + config: { user_id: 'owner-1' }, + contactCandidates: [{ id: 'c1', phone: '14155550123' }], + existingConversation: { id: 'cv1' }, + }); + const res = await resolveConversationByPhone( + db, + 'acct', + '+1 (415) 555-0123' + ); + expect(res).toEqual({ + conversationId: 'cv1', + contactId: 'c1', + contactCreated: false, + }); + }); + + it('creates contact + conversation when none exist', async () => { + const db = makeDb({ + config: { user_id: 'owner-1' }, + contactCandidates: [], + insertedContactId: 'c2', + existingConversation: null, + insertedConversationId: 'cv2', + }); + const res = await resolveConversationByPhone( + db, + 'acct', + '+14155550199', + 'Jane' + ); + expect(res).toEqual({ + conversationId: 'cv2', + contactId: 'c2', + contactCreated: true, + }); + }); + + it('re-resolves an existing contact when the insert loses a unique race', async () => { + // First lookup misses (→ we attempt an insert), the insert hits a + // 23505 unique violation, and the post-race re-lookup now returns + // the row a concurrent writer created. + const db = makeDb({ + config: { user_id: 'owner-1' }, + contactCandidatesByCall: [[], [{ id: 'c-raced', phone: '14155550123' }]], + insertContactError: { code: '23505' }, + existingConversation: { id: 'cv-raced' }, + }); + const res = await resolveConversationByPhone(db, 'acct', '+14155550123'); + expect(res.contactId).toBe('c-raced'); + expect(res.contactCreated).toBe(false); + expect(res.conversationId).toBe('cv-raced'); + }); +}); diff --git a/src/lib/whatsapp/resolve-conversation.ts b/src/lib/whatsapp/resolve-conversation.ts new file mode 100644 index 0000000..5789ea7 --- /dev/null +++ b/src/lib/whatsapp/resolve-conversation.ts @@ -0,0 +1,173 @@ +// ============================================================ +// Resolve (or create) the conversation for a phone number. +// +// The dashboard composer always has a `conversation_id` in hand. The +// public API doesn't — an external automation knows a *phone number*, +// not an internal UUID. This helper bridges that: given an E.164 +// phone, it finds-or-creates the contact and its conversation so the +// shared `sendMessageToConversation` core can run unchanged. +// +// It deliberately reuses the exact find-or-create logic the inbound +// webhook uses (the `findExistingContact` dedupe helper, the +// one-conversation-per-(account, contact) convention, the +// account_id-tenancy / user_id-audit split) so a contact created via +// the API is indistinguishable from one created by an inbound message. +// +// Audit user: created rows need a NOT NULL `user_id`. As with the +// webhook (where there's no logged-in human either), we attribute +// them to the WhatsApp config owner — a stable account-level default. +// ============================================================ + +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { findExistingContact, isUniqueViolation } from '@/lib/contacts/dedupe'; +import { sanitizePhoneForMeta, isValidE164 } from '@/lib/whatsapp/phone-utils'; +import { SendMessageError } from '@/lib/whatsapp/send-message'; +import { resolveAuditUserId, ContactError } from '@/lib/api/v1/contacts'; + +export interface ResolvedConversation { + conversationId: string; + contactId: string; + /** True if this call created the contact (vs matched an existing one). */ + contactCreated: boolean; +} + +/** + * Find or create the contact + conversation for `phone` within + * `accountId`. Throws `SendMessageError` (shared with the send core, + * so the route maps one error family) on a bad phone, a missing + * WhatsApp config, or a DB failure. + */ +export async function resolveConversationByPhone( + db: SupabaseClient, + accountId: string, + phone: string, + name?: string | null +): Promise { + const sanitized = sanitizePhoneForMeta(phone); + if (!isValidE164(sanitized)) { + throw new SendMessageError( + 'bad_request', + "'to' must be a valid phone number in E.164 format (e.g. +14155550123)", + 400 + ); + } + + // Fail fast (and create nothing) when the account has no WhatsApp + // connected — the same error the send would raise anyway. + const { data: config } = await db + .from('whatsapp_config') + .select('id') + .eq('account_id', accountId) + .maybeSingle(); + if (!config) { + throw new SendMessageError( + 'whatsapp_not_configured', + 'WhatsApp not configured. Please set up your WhatsApp integration first.', + 400 + ); + } + + // Audit user for created rows = the single account-wide default used + // by every public-API write (see resolveAuditUserId), so a contact + // created here is attributed identically to one created via + // POST /api/v1/contacts. resolveAuditUserId throws ContactError only + // if the owner can't be resolved — remap it to the send error family + // the callers already handle. + let ownerUserId: string; + try { + ownerUserId = await resolveAuditUserId(db, accountId); + } catch (err) { + if (err instanceof ContactError) { + throw new SendMessageError('db_error', err.message, err.status); + } + throw err; + } + + // ---- contact ------------------------------------------------- + let contactId: string; + let contactCreated = false; + + const existing = await findExistingContact(db, accountId, sanitized); + if (existing) { + contactId = existing.id; + if (name && name !== existing.name) { + await db + .from('contacts') + .update({ name, updated_at: new Date().toISOString() }) + .eq('id', existing.id); + } + } else { + const { data: created, error: createErr } = await db + .from('contacts') + .insert({ + account_id: accountId, + user_id: ownerUserId, + phone: sanitized, + name: name || sanitized, + }) + .select('id') + .single(); + + if (createErr || !created) { + // Lost a race against a concurrent inbound/API create — the + // unique index (migration 022) rejected the duplicate. Re-resolve. + if (isUniqueViolation(createErr)) { + const raced = await findExistingContact(db, accountId, sanitized); + if (raced) { + contactId = raced.id; + } else { + throw new SendMessageError( + 'db_error', + 'Failed to create contact', + 500 + ); + } + } else { + console.error( + '[resolve-conversation] contact create error:', + createErr + ); + throw new SendMessageError('db_error', 'Failed to create contact', 500); + } + } else { + contactId = created.id; + contactCreated = true; + } + } + + // ---- conversation ------------------------------------------- + // One conversation per (account, contact) — same convention as the + // webhook. + const { data: conv } = await db + .from('conversations') + .select('id') + .eq('account_id', accountId) + .eq('contact_id', contactId) + .maybeSingle(); + + if (conv?.id) { + return { conversationId: conv.id, contactId, contactCreated }; + } + + const { data: newConv, error: convErr } = await db + .from('conversations') + .insert({ + account_id: accountId, + user_id: ownerUserId, + contact_id: contactId, + }) + .select('id') + .single(); + + if (convErr || !newConv) { + console.error('[resolve-conversation] conversation create error:', convErr); + throw new SendMessageError( + 'db_error', + 'Failed to create conversation', + 500 + ); + } + + return { conversationId: newConv.id, contactId, contactCreated }; +} diff --git a/src/lib/whatsapp/send-message.test.ts b/src/lib/whatsapp/send-message.test.ts new file mode 100644 index 0000000..04550d7 --- /dev/null +++ b/src/lib/whatsapp/send-message.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { + sendMessageToConversation, + SendMessageError, + type SendMessageParams, +} from './send-message'; + +// A db that explodes if touched — these tests cover the param +// validation that MUST short-circuit before any query runs. +function noDb(): SupabaseClient { + return { + from() { + throw new Error('db should not be queried for invalid params'); + }, + } as unknown as SupabaseClient; +} + +async function expectSendError( + params: SendMessageParams, + status: number, + messageMatch?: RegExp +) { + await expect( + sendMessageToConversation(noDb(), 'acct-1', params) + ).rejects.toBeInstanceOf(SendMessageError); + await sendMessageToConversation(noDb(), 'acct-1', params).catch( + (e: SendMessageError) => { + expect(e.status).toBe(status); + if (messageMatch) expect(e.message).toMatch(messageMatch); + } + ); +} + +describe('sendMessageToConversation — param validation (pre-DB)', () => { + const base = { conversationId: 'cv-1' }; + + it('requires conversation_id and message_type', async () => { + await expectSendError({ conversationId: '', messageType: 'text' }, 400); + await expectSendError({ conversationId: 'cv-1', messageType: '' }, 400); + }); + + it('rejects an unsupported message_type', async () => { + await expectSendError( + { ...base, messageType: 'carrier-pigeon' }, + 400, + /Unsupported message_type/ + ); + }); + + it('requires content_text for text messages', async () => { + await expectSendError( + { ...base, messageType: 'text' }, + 400, + /content_text is required/ + ); + }); + + it('requires template_name for template messages', async () => { + await expectSendError( + { ...base, messageType: 'template' }, + 400, + /template_name is required/ + ); + }); + + it('requires media_url for media kinds', async () => { + for (const kind of ['image', 'video', 'document', 'audio']) { + await expectSendError( + { ...base, messageType: kind }, + 400, + /media_url is required/ + ); + } + }); + + it('rejects an over-long media caption (non-audio)', async () => { + await expectSendError( + { + ...base, + messageType: 'image', + mediaUrl: 'https://x/y.jpg', + contentText: 'a'.repeat(1025), + }, + 400, + /1024-character limit/ + ); + }); + + it('allows a long "caption" on audio (audio carries none) — so it reaches the DB', async () => { + // Audio is exempt from the caption cap, so validation passes and we + // proceed to the conversation lookup — proven by the stub throwing. + const spy = vi.fn(() => { + throw new Error('reached DB'); + }); + const db = { from: spy } as unknown as SupabaseClient; + await expect( + sendMessageToConversation(db, 'acct-1', { + ...base, + messageType: 'audio', + mediaUrl: 'https://x/y.ogg', + contentText: 'a'.repeat(2000), + }) + ).rejects.toThrow('reached DB'); + expect(spy).toHaveBeenCalledWith('conversations'); + }); +}); + +describe('SendMessageError', () => { + it('carries a machine code and an HTTP status', () => { + const e = new SendMessageError('meta_error', 'boom', 502); + expect(e.code).toBe('meta_error'); + expect(e.status).toBe(502); + expect(e).toBeInstanceOf(Error); + }); +}); diff --git a/src/lib/whatsapp/send-message.ts b/src/lib/whatsapp/send-message.ts new file mode 100644 index 0000000..8637955 --- /dev/null +++ b/src/lib/whatsapp/send-message.ts @@ -0,0 +1,447 @@ +// ============================================================ +// Outbound message send — the core that both the dashboard's +// `/api/whatsapp/send` route and the public `/api/v1/messages` +// endpoint call. +// +// Given a conversation and message params, this: +// 1. validates the params for the message type, +// 2. loads the conversation + contact + WhatsApp config, +// 3. sends to Meta (with phone-variant retry + contact auto-fix), +// 4. persists the message + updates the conversation, +// 5. pauses any active Flow run for the contact (agent stepped in). +// +// It is transport-agnostic: it takes a `SupabaseClient` and an +// `accountId` and throws `SendMessageError` on failure. The callers +// own auth, rate-limiting, body parsing, and mapping the error to +// their respective response shapes (internal `{ error }` vs the v1 +// envelope). Behaviour is identical to the original inline route — +// this is a straight extraction so the public endpoint can reuse it +// without duplicating ~250 lines of Meta plumbing. +// ============================================================ + +import type { SupabaseClient } from '@supabase/supabase-js'; + +import { + sendTextMessage, + sendTemplateMessage, + sendMediaMessage, + type MediaKind, +} from '@/lib/whatsapp/meta-api'; +import { decrypt, encrypt, isLegacyFormat } from '@/lib/whatsapp/encryption'; +import { supabaseAdmin } from '@/lib/flows/admin-client'; +import { + sanitizePhoneForMeta, + isValidE164, + phoneVariants, + isRecipientNotAllowedError, +} from '@/lib/whatsapp/phone-utils'; +import type { MessageTemplate } from '@/types'; +import { isMessageTemplate } from '@/lib/whatsapp/template-row-guard'; + +export const MEDIA_KINDS = ['image', 'video', 'document', 'audio'] as const; +export const VALID_MESSAGE_TYPES = [ + 'text', + 'template', + ...MEDIA_KINDS, +] as const; + +/** + * Typed failure with a machine `code` and a suggested HTTP `status`. + * Callers map it to their own response shape (`toErrorResponse` for + * the dashboard route, the v1 envelope for the public endpoint). + */ +export class SendMessageError extends Error { + readonly code: string; + readonly status: number; + constructor(code: string, message: string, status: number) { + super(message); + this.name = 'SendMessageError'; + this.code = code; + this.status = status; + } +} + +export interface SendMessageParams { + conversationId: string; + messageType: string; + contentText?: string | null; + mediaUrl?: string | null; + filename?: string | null; + templateName?: string | null; + templateLanguage?: string | null; + /** Legacy positional body params (only used if messageParams.body unset). */ + templateParams?: string[]; + /** Structured template params (header/body/buttons). */ + templateMessageParams?: unknown; + replyToMessageId?: string | null; +} + +export interface SendMessageResult { + /** Our `messages.id` (the persisted row). */ + messageId: string; + /** Meta's `wamid` for the delivered message. */ + whatsappMessageId: string; +} + +/** + * Send a message in an existing conversation and persist it. + * + * `db` may be an RLS-scoped user client (dashboard) or the service- + * role client (public API) — every query is filtered by `accountId` + * either way, so tenancy holds regardless of which client is passed. + */ +/** + * Validate the message-shape params (type, required content, caption + * cap) independently of any DB state, throwing `SendMessageError` on a + * bad payload. Exported so a caller can reject a malformed request + * *before* it finds-or-creates a contact/conversation — otherwise an + * invalid payload leaves an orphan empty conversation behind. The send + * core calls this too, so validation can't be skipped. + */ +export function validateSendMessageParams(params: { + messageType: string; + contentText?: string | null; + mediaUrl?: string | null; + templateName?: string | null; +}): void { + const { messageType, contentText, mediaUrl, templateName } = params; + + if (!messageType) { + throw new SendMessageError('bad_request', 'message_type is required', 400); + } + + const isMediaKind = (MEDIA_KINDS as readonly string[]).includes(messageType); + + if (!(VALID_MESSAGE_TYPES as readonly string[]).includes(messageType)) { + throw new SendMessageError( + 'bad_request', + `Unsupported message_type "${messageType}"`, + 400 + ); + } + + if (messageType === 'text' && !contentText) { + throw new SendMessageError( + 'bad_request', + 'content_text is required for text messages', + 400 + ); + } + + if (messageType === 'template' && !templateName) { + throw new SendMessageError( + 'bad_request', + 'template_name is required for template messages', + 400 + ); + } + + if (isMediaKind && !mediaUrl) { + throw new SendMessageError( + 'bad_request', + `media_url is required for ${messageType} messages`, + 400 + ); + } + + // Meta caps media captions at 1024 chars (audio carries none). + if ( + isMediaKind && + messageType !== 'audio' && + typeof contentText === 'string' && + contentText.length > 1024 + ) { + throw new SendMessageError( + 'bad_request', + 'Caption exceeds the 1024-character limit', + 400 + ); + } +} + +export async function sendMessageToConversation( + db: SupabaseClient, + accountId: string, + params: SendMessageParams +): Promise { + const { + conversationId, + messageType, + contentText, + mediaUrl, + filename, + templateName, + templateLanguage, + templateParams, + templateMessageParams, + replyToMessageId, + } = params; + + if (!conversationId) { + throw new SendMessageError( + 'bad_request', + 'conversation_id is required', + 400 + ); + } + + validateSendMessageParams({ messageType, contentText, mediaUrl, templateName }); + + const isMediaKind = (MEDIA_KINDS as readonly string[]).includes(messageType); + + // Conversation + contact, account-scoped. + const { data: conversation, error: convError } = await db + .from('conversations') + .select('*, contact:contacts(*)') + .eq('id', conversationId) + .eq('account_id', accountId) + .single(); + + if (convError || !conversation) { + throw new SendMessageError('not_found', 'Conversation not found', 404); + } + + const contact = conversation.contact; + if (!contact?.phone) { + throw new SendMessageError( + 'bad_request', + 'Contact phone number not found', + 400 + ); + } + + const sanitizedPhone = sanitizePhoneForMeta(contact.phone); + if (!isValidE164(sanitizedPhone)) { + throw new SendMessageError( + 'bad_request', + 'Invalid phone number format', + 400 + ); + } + + // WhatsApp config, account-scoped. + const { data: config, error: configError } = await db + .from('whatsapp_config') + .select('*') + .eq('account_id', accountId) + .single(); + + if (configError || !config) { + throw new SendMessageError( + 'whatsapp_not_configured', + 'WhatsApp not configured. Please set up your WhatsApp integration first.', + 400 + ); + } + + const accessToken = decrypt(config.access_token); + + // Self-heal legacy CBC ciphertexts. Fire-and-forget; idempotent. + if (isLegacyFormat(config.access_token)) { + void db + .from('whatsapp_config') + .update({ access_token: encrypt(accessToken) }) + .eq('id', config.id) + .then(({ error }: { error: { message: string } | null }) => { + if (error) { + console.warn( + '[send-message] access_token GCM upgrade failed:', + error.message + ); + } + }); + } + + // Resolve the reply target to its Meta message_id. The parent must + // belong to this same conversation — otherwise a caller could quote + // messages they can't see by guessing UUIDs. + let contextMessageId: string | undefined; + if (replyToMessageId) { + const { data: parent, error: parentError } = await db + .from('messages') + .select('message_id, conversation_id') + .eq('id', replyToMessageId) + .eq('conversation_id', conversationId) + .maybeSingle(); + + if (parentError || !parent) { + throw new SendMessageError( + 'bad_request', + 'reply_to_message_id not found in this conversation', + 400 + ); + } + if (!parent.message_id) { + console.warn( + '[send-message] reply target has no Meta message_id; sending without context' + ); + } else { + contextMessageId = parent.message_id; + } + } + + // Template row (for header + button components). isMessageTemplate + // guards against a malformed local row crashing the send-builder. + let templateRow: MessageTemplate | null = null; + if (messageType === 'template' && templateName) { + const { data } = await db + .from('message_templates') + .select('*') + .eq('account_id', accountId) + .eq('name', templateName) + .eq('language', templateLanguage || 'en_US') + .maybeSingle(); + if (data && !isMessageTemplate(data)) { + throw new SendMessageError( + 'template_malformed', + 'Template row is malformed locally — run "Sync from Meta" in Settings to repair it.', + 500 + ); + } + templateRow = data ?? null; + } + + const attempt = async (phone: string): Promise => { + if (messageType === 'template') { + const result = await sendTemplateMessage({ + phoneNumberId: config.phone_number_id, + accessToken, + to: phone, + templateName: templateName!, + language: templateLanguage || 'en_US', + template: templateRow ?? undefined, + messageParams: templateMessageParams ?? undefined, + params: templateParams || [], + contextMessageId, + }); + return result.messageId; + } + if (isMediaKind) { + const result = await sendMediaMessage({ + phoneNumberId: config.phone_number_id, + accessToken, + to: phone, + kind: messageType as MediaKind, + link: mediaUrl!, + caption: contentText || undefined, + filename: filename || undefined, + contextMessageId, + }); + return result.messageId; + } + const result = await sendTextMessage({ + phoneNumberId: config.phone_number_id, + accessToken, + to: phone, + text: contentText!, + contextMessageId, + }); + return result.messageId; + }; + + // Send via Meta — retry across phone-number variants if Meta rejects + // with "recipient not in allowed list"; persist a working variant + // back to the contact so the next send goes straight through. + let waMessageId = ''; + let workingPhone = sanitizedPhone; + try { + const variants = phoneVariants(sanitizedPhone); + let lastError: unknown = null; + + for (const variant of variants) { + try { + waMessageId = await attempt(variant); + workingPhone = variant; + lastError = null; + break; + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + if (!isRecipientNotAllowedError(message)) { + throw err; + } + lastError = err; + console.warn( + `[send-message] variant "${variant}" rejected by Meta, trying next…` + ); + } + } + + if (lastError) throw lastError; + } catch (err) { + const message = + err instanceof Error ? err.message : 'Unknown Meta API error'; + console.error('[send-message] Meta send failed for all variants:', message); + throw new SendMessageError('meta_error', `Meta API error: ${message}`, 502); + } + + if (workingPhone !== sanitizedPhone) { + console.log( + `[send-message] Auto-corrected contact phone: ${sanitizedPhone} → ${workingPhone}` + ); + await db + .from('contacts') + .update({ phone: workingPhone }) + .eq('id', contact.id); + } + + // Persist the sent message. Field names MUST match the messages + // schema (see 001_initial_schema.sql). + const { data: messageRecord, error: msgError } = await db + .from('messages') + .insert({ + conversation_id: conversationId, + sender_type: 'agent', + content_type: messageType, + content_text: contentText || null, + media_url: mediaUrl || null, + template_name: templateName || null, + message_id: waMessageId, + status: 'sent', + reply_to_message_id: replyToMessageId || null, + }) + .select() + .single(); + + if (msgError) { + console.error('[send-message] error inserting sent message:', msgError); + throw new SendMessageError( + 'db_error', + `Message sent to Meta but failed to save to DB: ${msgError.message}`, + 500 + ); + } + + await db + .from('conversations') + .update({ + last_message_text: contentText || `[${messageType}]`, + last_message_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + }) + .eq('id', conversationId); + + // Pause any active Flow run for this contact — the agent stepping in + // is the strongest "yield, human is here" signal. Best-effort. + try { + const { error: pauseErr } = await supabaseAdmin() + .from('flow_runs') + .update({ + status: 'paused_by_agent', + ended_at: new Date().toISOString(), + end_reason: 'agent_replied', + }) + .eq('account_id', accountId) + .eq('contact_id', contact.id) + .eq('status', 'active'); + if (pauseErr) { + console.error('[flows] pause-on-agent-send failed:', pauseErr.message); + } + } catch (err) { + console.error( + '[flows] pause-on-agent-send threw:', + err instanceof Error ? err.message : err + ); + } + + return { messageId: messageRecord.id, whatsappMessageId: waMessageId }; +}