feat(api): public REST API v1 — messages, contacts, conversations, broadcasts (#245)

Complete the public API roadmap on top of the existing key-auth layer
(#290). Re-lands POST /api/v1/messages (lost when the stacked #291 never
reached main) by extracting the dashboard send route's core into
`sendMessageToConversation`, then adds:

- Contacts: GET/POST /api/v1/contacts, GET/PATCH /api/v1/contacts/{id}
- Conversations: GET list + /{id} + /{id}/messages
- Broadcasts: POST /api/v1/broadcasts + GET /{id} (persist rows, fan out
  in after(), poll for status)

All list endpoints share one keyset-pagination contract
({ data, meta: { next_cursor } }). No migration required — scopes and
tables already exist. Versioned as 0.3.0 (folded into [Unreleased]);
docs/public-api.md updated; outbound webhooks left as the sole roadmap
item. API stays v1 (additive).

Includes fixes from an adversarial code review: cursor validation
against filter injection, broadcast counts deferred to the DB aggregate
trigger, terminal broadcast status reflects failures, duplicate-phone
de-dup, up-front send-param validation (no orphan conversations),
PATCH null-clear semantics, error-checked tag sync, bounded tag filter,
unified audit-user attribution, and a maxDuration on the broadcast
fan-out.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Arnas Donauskas 2026-07-01 14:10:19 +03:00
parent 41815acc88
commit 8cb5792823
26 changed files with 2968 additions and 381 deletions

View File

@ -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 <key>`. 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

View File

@ -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": "<uuid>" // 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=<tagId>`.
```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.

View File

@ -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",

View File

@ -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);
}
}

View File

@ -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);
}
}

View File

@ -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<string, unknown> = {};
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);
}
}

View File

@ -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=<tagId>` 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<string, unknown>)),
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);
}
}

View File

@ -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);
}
}

View File

@ -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);
}
}

View File

@ -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);
}
}

View File

@ -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": "<uuid>", // 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<string, unknown>)
: 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);
}
}

View File

@ -14,6 +14,10 @@ const messageInserts: Array<Record<string, unknown>> = []
// Toggles for the per-test scenario.
let existingConversation: Record<string, unknown> | null = null
let contactRow: Record<string, unknown> | 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<string, unknown> | 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<string, unknown>) => {
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()

View File

@ -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<string> => {
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<ReturnType<typeof createClient>>
/**
* 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<string | null> {
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
}

View File

@ -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);
});
});

223
src/lib/api/v1/contacts.ts Normal file
View File

@ -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<string, unknown>): 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<RawTagJoin['tags']> => 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<string> {
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<void> {
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<ApiContact | null> {
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<string, unknown>);
}

View File

@ -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');
});
});

View File

@ -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,
};
}

View File

@ -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]));
});
});

View File

@ -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<T extends { created_at: string; id: string }>(
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]) };
}

View File

@ -86,6 +86,33 @@ export function ok<T>(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<T>(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<string, string>
): 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

View File

@ -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 });
});
});

View File

@ -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<BroadcastPlan> {
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<string>();
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<void> {
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);
}

View File

@ -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<string, unknown> = {
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');
});
});

View File

@ -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<ResolvedConversation> {
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 };
}

View File

@ -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);
});
});

View File

@ -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<SendMessageResult> {
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<string> => {
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 };
}