LaLeV/wacrm
1
0
Fork 0
mirror of https://github.com/Luchexx/wacrm.git synced 2026-11-02 09:19:51 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(api): outbound event webhooks — /api/v1/webhooks (#245)

Browse Source 
Completes the public API roadmap (#245): register HTTPS endpoints to be
POSTed to when things happen in an account, so automations can react to
inbound activity instead of polling.

- Migration 028: account-scoped webhook_endpoints (secret stored
  AES-GCM-encrypted) + an atomic record_webhook_failure() function.
- Scope webhooks:manage; management API GET/POST /api/v1/webhooks and
  GET/PATCH/DELETE /api/v1/webhooks/{id} (secret returned once).
- Events message.received / conversation.created / message.status_updated
  dispatched from the inbound webhook handler's after() block.
- Signed delivery (X-Wacrm-Signature = HMAC-SHA256 over t.body), single
  best-effort attempt, auto-disable after consecutive failures.
- Versioned 0.4.0; docs/public-api.md Webhooks section; CHANGELOG.

Fixes from the adversarial code review of this diff:
- SSRF: block private/loopback/link-local (incl. cloud metadata) targets
  at delivery time + redirect:'manual' so a public URL can't bounce
  internal (src/lib/webhooks/ssrf.ts).
- Status mirror UPDATE no longer uses .select().maybeSingle() (message_id
  is non-unique); account resolved via a bounded limit(1) read.
- Atomic failure counter via SQL RPC (was a racy read-modify-write).
- conversation.created emitted before the reaction short-circuit.
- Per-delivery id added for receiver dedupe; docs state at-least-once /
  unordered delivery.
- Signature verify tolerates hex case / header whitespace.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Arnas Donauskas 2026-07-01 14:53:18 +03:00
parent 9f76468a41
commit c0f390c401
18 changed files with 1314 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGELOG.md
View File

@ -9,6 +9,28 @@ 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.
## [0.4.0] — 2026-07-01
Completes the public API (#245): **outbound event webhooks** so
automations can *react* to activity instead of polling.
### Added
- **Outbound event webhooks (`/api/v1/webhooks`).** Register an HTTPS
endpoint (scope `webhooks:manage`) to be POSTed to when an event
happens in your account — `message.received`, `message.status_updated`,
or `conversation.created`. Manage endpoints with
`GET/POST /api/v1/webhooks` and `GET/PATCH/DELETE /api/v1/webhooks/{id}`.
Each delivery is signed with an `X-Wacrm-Signature`
(HMAC-SHA256 over `timestamp.body`) so receivers can verify
authenticity and reject replays; the signing secret is returned once
at creation and stored encrypted. Delivery is best-effort — an
endpoint that fails repeatedly is auto-disabled after a threshold of
consecutive failures. See `docs/public-api.md`.
**Migration required:** apply
`supabase/migrations/028_webhook_endpoints.sql`.
([#245](https://github.com/ArnasDon/wacrm/issues/245))
## [0.3.0] — 2026-07-01
Multi-user accounts ship. Every wacrm install is multi-tenant on the

116
docs/public-api.md
View File

@ -4,10 +4,9 @@ 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:** 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).
> **Status:** stable. Authentication, scopes, rate limiting, the
> messages / contacts / conversations / broadcasts endpoints, and
> outbound event [webhooks](#webhooks) all ship now.
## Authentication
@ -50,6 +49,7 @@ it. Grant the minimum.
| `contacts:write` | Create and update contacts |
| `conversations:read` | List and read conversations |
| `broadcasts:send` | Launch broadcast campaigns |
| `webhooks:manage` | Register and manage outbound webhooks |
A key with **no scopes** still authenticates and can call
`GET /api/v1/me` — useful for verifying a key works.
@ -281,15 +281,103 @@ Cursors are keyset-based (stable under concurrent inserts). Pass the
cursor back verbatim — don't parse it. `next_cursor: null` means the
last page.
## Webhooks
Rather than polling, register an endpoint and wacrm will POST to it when
things happen in your account. **Migration required:** apply
`supabase/migrations/028_webhook_endpoints.sql`.
### Events
| Event | Fires when |
| ------------------------ | ------------------------------------------------- |
| `message.received` | An inbound message arrives from a contact |
| `message.status_updated` | A message you sent changed delivery status |
| `conversation.created` | A new conversation is opened for a contact |
### Managing endpoints
All under scope `webhooks:manage`.
- `POST /api/v1/webhooks` — register `{ "url": "https://…", "events": ["message.received"] }`. `url` must be `https://`. **The response includes `secret` exactly once** — store it to verify signatures; wacrm keeps only an encrypted copy.
- `GET /api/v1/webhooks` — list your endpoints (never returns the secret).
- `GET /api/v1/webhooks/{id}` — read one.
- `PATCH /api/v1/webhooks/{id}` — update `url`, `events`, or `is_active` (re-enabling clears the failure counter).
- `DELETE /api/v1/webhooks/{id}` — remove one.
```bash
curl -X POST https://your-crm.example.com/api/v1/webhooks \
-H "Authorization: Bearer wacrm_live_xxx" \
-H "Content-Type: application/json" \
-d '{ "url": "https://example.com/hooks/wacrm", "events": ["message.received"] }'
# → 201 { "data": { "id": "…", "url": "…", "events": [...], "secret": "whsec_…" } }
```
### Delivery payload
Every delivery is a POST with this envelope; `id` is a unique per-
delivery uuid you can dedupe on, and `data` varies by `event`:
```json
{
"id": "8f3c…",
"event": "message.received",
"occurred_at": "2026-07-01T12:00:00.000Z",
"account_id": "…",
"data": { /* per-event, see below */ }
}
```
`data` by event:
```jsonc
// message.received
{ "conversation_id": "…", "contact_id": "…", "whatsapp_message_id": "wamid.…", "content_type": "text", "text": "Hi 👋" }
// conversation.created
{ "conversation_id": "…", "contact_id": "…" }
// message.status_updated
{ "whatsapp_message_id": "wamid.…", "conversation_id": "…", "status": "delivered" }
```
Headers: `X-Wacrm-Event`, `X-Wacrm-Webhook-Id`, and `X-Wacrm-Signature`.
### Verifying the signature
`X-Wacrm-Signature: t=<unix_seconds>,v1=<hex>` where `v1 =
HMAC-SHA256(secret, "${t}.${rawBody}")`. Recompute it over the **raw
request body** and compare in constant time; reject if `t` is more than
a few minutes old (replay protection).
```js
const [, t, v1] = header.match(/t=(\d+),v1=([0-9a-f]+)/);
const expected = crypto.createHmac('sha256', secret)
.update(`${t}.${rawBody}`).digest('hex');
const ok = crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(v1));
```
### Delivery semantics
Delivery is **best-effort**: a single attempt per event with a short
timeout, and **redirects are not followed**. `message.status_updated`
covers messages wacrm stores (inbox + API sends), not broadcast-only
sends, and — because providers re-send and re-order status callbacks —
the same status may arrive more than once or out of order; **dedupe on
`id` and don't assume ordering**. Each consecutive failure increments
`failure_count`; after enough consecutive failures the endpoint is
auto-disabled (`is_active: false`) — re-enable it with `PATCH` (which
resets the counter). Durable retry-with-backoff (a delivery queue) is a
future enhancement; today, treat missed deliveries as possible and
reconcile with the read endpoints when it matters.
**Target restrictions (SSRF).** The `url` must be `https://` and must
resolve to a public address — requests to `localhost`, private/RFC1918
ranges, link-local (incl. cloud metadata `169.254.169.254`), and similar
internal targets are refused at delivery time.
## Roadmap
The one remaining item tracked in
[#245](https://github.com/ArnasDon/wacrm/issues/245):
- **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.
The public API now covers messaging, contacts, conversations,
broadcasts, and outbound webhooks — the full scope of
[#245](https://github.com/ArnasDon/wacrm/issues/245). Future ideas
(deals/pipelines, templates, flows, a delivery queue for webhooks) are
not yet scheduled.

2
package.json
View File

@ -1,6 +1,6 @@
{
"name": "wacrm",
"version": "0.3.0",
"version": "0.4.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",

146
src/app/api/v1/webhooks/[id]/route.ts Normal file
View File

@ -0,0 +1,146 @@
// ============================================================
// GET /api/v1/webhooks/{id} — read an endpoint (webhooks:manage)
// PATCH /api/v1/webhooks/{id} — update url/events/is_active
// DELETE /api/v1/webhooks/{id} — remove an endpoint
//
// All account-scoped: a foreign id → 404 (never 403). The signing
// secret is never returned here — it's shown once at creation only.
// ============================================================
import { requireApiKey } from '@/lib/auth/api-context';
import { ok, fail, toApiErrorResponse } from '@/lib/api/v1/respond';
import { normalizeEvents } from '@/lib/webhooks/events';
import {
WEBHOOK_PUBLIC_COLUMNS,
serializeWebhookEndpoint,
normalizeWebhookUrl,
} from '@/lib/webhooks/endpoints';
export async function GET(
request: Request,
{ params }: { params: Promise<{ id: string }> }
) {
try {
const ctx = await requireApiKey(request, 'webhooks:manage');
const { id } = await params;
const { data, error } = await ctx.supabase
.from('webhook_endpoints')
.select(WEBHOOK_PUBLIC_COLUMNS)
.eq('id', id)
.eq('account_id', ctx.accountId)
.maybeSingle();
if (error) {
console.error('[api/v1/webhooks] read error:', error);
return fail('internal', 'Failed to read webhook', 500);
}
if (!data) return fail('not_found', 'Webhook not found', 404);
return ok(serializeWebhookEndpoint(data as Record<string, unknown>));
} catch (err) {
return toApiErrorResponse(err);
}
}
export async function PATCH(
request: Request,
{ params }: { params: Promise<{ id: string }> }
) {
try {
const ctx = await requireApiKey(request, 'webhooks:manage');
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);
}
const updates: Record<string, unknown> = {};
if ('url' in body) {
const url = normalizeWebhookUrl(body.url);
if (!url) {
return fail('bad_request', "'url' must be a valid https:// URL", 400);
}
updates.url = url;
}
if ('events' in body) {
const events = normalizeEvents(body.events);
if (!events) {
return fail(
'bad_request',
"'events' must be a non-empty array of known event names",
400
);
}
updates.events = events;
}
if ('is_active' in body) {
if (typeof body.is_active !== 'boolean') {
return fail('bad_request', "'is_active' must be a boolean", 400);
}
updates.is_active = body.is_active;
// Re-enabling a disabled endpoint clears its failure streak so it
// isn't instantly re-disabled by a single stale failure.
if (body.is_active === true) updates.failure_count = 0;
}
if (Object.keys(updates).length === 0) {
return fail('bad_request', 'No updatable fields provided', 400);
}
// Scope the update by account_id so a foreign id touches nothing;
// the returned row (null when unmatched) drives the 404.
const { data, error } = await ctx.supabase
.from('webhook_endpoints')
.update(updates)
.eq('id', id)
.eq('account_id', ctx.accountId)
.select(WEBHOOK_PUBLIC_COLUMNS)
.maybeSingle();
if (error) {
console.error('[api/v1/webhooks] update error:', error);
return fail('internal', 'Failed to update webhook', 500);
}
if (!data) return fail('not_found', 'Webhook not found', 404);
return ok(serializeWebhookEndpoint(data as Record<string, unknown>));
} catch (err) {
return toApiErrorResponse(err);
}
}
export async function DELETE(
request: Request,
{ params }: { params: Promise<{ id: string }> }
) {
try {
const ctx = await requireApiKey(request, 'webhooks:manage');
const { id } = await params;
const { data, error } = await ctx.supabase
.from('webhook_endpoints')
.delete()
.eq('id', id)
.eq('account_id', ctx.accountId)
.select('id')
.maybeSingle();
if (error) {
console.error('[api/v1/webhooks] delete error:', error);
return fail('internal', 'Failed to delete webhook', 500);
}
if (!data) return fail('not_found', 'Webhook not found', 404);
return ok({ id: data.id, deleted: true });
} catch (err) {
return toApiErrorResponse(err);
}
}

102
src/app/api/v1/webhooks/route.ts Normal file
View File

@ -0,0 +1,102 @@
// ============================================================
// GET /api/v1/webhooks — list webhook endpoints (scope: webhooks:manage)
// POST /api/v1/webhooks — register an endpoint (scope: webhooks:manage)
//
// POST returns the signing `secret` in plaintext exactly once — store
// it to verify the `X-Wacrm-Signature` on deliveries. wacrm keeps only
// an encrypted copy and can never show it again.
// ============================================================
import { requireApiKey } from '@/lib/auth/api-context';
import { ok, okList, fail, toApiErrorResponse } from '@/lib/api/v1/respond';
import { encrypt } from '@/lib/whatsapp/encryption';
import { normalizeEvents } from '@/lib/webhooks/events';
import {
WEBHOOK_PUBLIC_COLUMNS,
serializeWebhookEndpoint,
generateWebhookSecret,
normalizeWebhookUrl,
} from '@/lib/webhooks/endpoints';
export async function GET(request: Request) {
try {
const ctx = await requireApiKey(request, 'webhooks:manage');
const { data, error } = await ctx.supabase
.from('webhook_endpoints')
.select(WEBHOOK_PUBLIC_COLUMNS)
.eq('account_id', ctx.accountId)
.order('created_at', { ascending: false });
if (error) {
console.error('[api/v1/webhooks] list error:', error);
return fail('internal', 'Failed to list webhooks', 500);
}
// The roster is small and settings-class — return it whole (the
// list envelope's cursor is always null here).
return okList(
(data ?? []).map((r) =>
serializeWebhookEndpoint(r as Record<string, unknown>)
),
null
);
} catch (err) {
return toApiErrorResponse(err);
}
}
export async function POST(request: Request) {
try {
const ctx = await requireApiKey(request, 'webhooks:manage');
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 url = normalizeWebhookUrl(body.url);
if (!url) {
return fail('bad_request', "'url' must be a valid https:// URL", 400);
}
const events = normalizeEvents(body.events);
if (!events) {
return fail(
'bad_request',
"'events' must be a non-empty array of known event names",
400
);
}
const secret = generateWebhookSecret();
const { data: created, error } = await ctx.supabase
.from('webhook_endpoints')
.insert({
account_id: ctx.accountId,
created_by: ctx.createdBy,
url,
secret: encrypt(secret),
events,
})
.select(WEBHOOK_PUBLIC_COLUMNS)
.single();
if (error || !created) {
console.error('[api/v1/webhooks] create error:', error);
return fail('internal', 'Failed to create webhook', 500);
}
// Secret shown exactly once.
return ok(
{ ...serializeWebhookEndpoint(created as Record<string, unknown>), secret },
201
);
} catch (err) {
return toApiErrorResponse(err);
}
}

106
src/app/api/whatsapp/webhook/route.ts
View File

@ -7,6 +7,7 @@ import { findExistingContact, isUniqueViolation } from '@/lib/contacts/dedupe'
import { verifyMetaWebhookSignature } from '@/lib/whatsapp/webhook-signature'
import { runAutomationsForTrigger } from '@/lib/automations/engine'
import { dispatchInboundToFlows } from '@/lib/flows/engine'
import { dispatchWebhookEvent } from '@/lib/webhooks/deliver'
import {
handleTemplateWebhookChange,
isTemplateWebhookField,
@ -354,7 +355,10 @@ async function handleStatusUpdate(status: {
recipient_id: string
}) {
// 1) Mirror onto messages (legacy behavior) — Meta's status values
// already match the CHECK constraint on messages.status.
// already match the CHECK constraint on messages.status. No
// `.select()`: message_id is NOT unique (migration 009 — Meta ids
// repeat across numbers), so this updates 0..N rows and must not
// assume a single row.
const { error: msgErr } = await supabaseAdmin()
.from('messages')
.update({ status: status.status })
@ -364,6 +368,10 @@ async function handleStatusUpdate(status: {
console.error('Error updating message status:', msgErr)
}
// Webhook fan-out for this status change happens at the END of this
// handler (after the broadcast mirror below), so a slow subscriber
// endpoint can't delay the broadcast_recipients update.
// 2) Mirror onto broadcast_recipients via whatsapp_message_id
// (added in migration 003). The aggregate trigger on
// broadcast_recipients re-derives the parent broadcast's
@ -378,26 +386,53 @@ async function handleStatusUpdate(status: {
if (recFetchErr) {
console.error('Error fetching broadcast recipient:', recFetchErr)
return
} else if (
recipient &&
// Guard transitions — forward-only on the success ladder, and
// `failed` only from pre-delivered states.
isValidStatusTransition(recipient.status, status.status)
) {
const update: Record<string, unknown> = { status: status.status }
if (status.status === 'sent' && !('sent_at' in update)) update.sent_at = tsIso
if (status.status === 'delivered') update.delivered_at = tsIso
if (status.status === 'read') update.read_at = tsIso
const { error: recUpdateErr } = await supabaseAdmin()
.from('broadcast_recipients')
.update(update)
.eq('id', recipient.id)
if (recUpdateErr) {
console.error('Error updating broadcast recipient status:', recUpdateErr)
}
}
if (!recipient) return // message wasn't part of a broadcast — fine
// Guard transitions — forward-only on the success ladder, and
// `failed` only from pre-delivered states.
if (!isValidStatusTransition(recipient.status, status.status)) return
// 3) Webhook fan-out for messages we store (inbox / API sends).
// Runs last so a slow subscriber can't delay the mirrors above.
// Bounded to one row (message_id isn't unique) purely to resolve
// the owning account for delivery.
const { data: msgRow } = await supabaseAdmin()
.from('messages')
.select('conversation_id, conversations(account_id)')
.eq('message_id', status.id)
.limit(1)
.maybeSingle()
const update: Record<string, unknown> = { status: status.status }
if (status.status === 'sent' && !('sent_at' in update)) update.sent_at = tsIso
if (status.status === 'delivered') update.delivered_at = tsIso
if (status.status === 'read') update.read_at = tsIso
const { error: recUpdateErr } = await supabaseAdmin()
.from('broadcast_recipients')
.update(update)
.eq('id', recipient.id)
if (recUpdateErr) {
console.error('Error updating broadcast recipient status:', recUpdateErr)
if (msgRow) {
const conv = msgRow.conversations as { account_id: string } | null
const accountId = conv?.account_id
if (accountId) {
await dispatchWebhookEvent(
supabaseAdmin(),
accountId,
'message.status_updated',
{
whatsapp_message_id: status.id,
conversation_id: msgRow.conversation_id,
status: status.status,
}
)
}
}
}
@ -548,12 +583,24 @@ async function processMessage(
const contactRecord = contactOutcome.contact
// Find or create conversation
const conversation = await findOrCreateConversation(
const convResult = await findOrCreateConversation(
accountId,
configOwnerUserId,
contactRecord.id
)
if (!conversation) return
if (!convResult) return
const conversation = convResult.conversation
// Emit conversation.created as soon as the thread is opened — BEFORE
// the reaction short-circuit below — so a conversation first opened by
// a reaction still fires the event, and a subscriber always sees the
// thread open before its first message.received.
if (convResult.created) {
await dispatchWebhookEvent(supabaseAdmin(), accountId, 'conversation.created', {
conversation_id: conversation.id,
contact_id: contactRecord.id,
})
}
// Reactions short-circuit here — they aren't messages. We never insert
// into `messages`, never bump unread_count, never update last_message_text.
@ -736,6 +783,21 @@ async function processMessage(
},
}).catch((err) => console.error('[automations] dispatch failed:', err))
}
// message.received webhook (public API). Awaited — not fire-and-forget
// — because we're inside the route's `after()` block, which only keeps
// the function alive for promises it can see; a detached promise could
// be frozen before it delivers. `dispatchWebhookEvent` early-exits
// when the account has no matching endpoint and never throws.
// (conversation.created is emitted earlier, right after the thread is
// opened.)
await dispatchWebhookEvent(supabaseAdmin(), accountId, 'message.received', {
conversation_id: conversation.id,
contact_id: contactRecord.id,
whatsapp_message_id: message.id,
content_type: contentType,
text: contentText,
})
}
async function parseMessageContent(
@ -967,7 +1029,7 @@ async function findOrCreateConversation(
.single()
if (!findError && existing) {
return existing
return { conversation: existing, created: false }
}
// Create new conversation. Same tenancy + audit split as
@ -987,5 +1049,5 @@ async function findOrCreateConversation(
return null
}
return newConv
return { conversation: newConv, created: true }
}

2
src/lib/api-keys/scopes.ts
View File

@ -20,6 +20,7 @@ export const API_SCOPES = [
'contacts:write',
'conversations:read',
'broadcasts:send',
'webhooks:manage',
] as const;
export type ApiScope = (typeof API_SCOPES)[number];
@ -32,6 +33,7 @@ export const SCOPE_DESCRIPTIONS: Record<ApiScope, string> = {
'contacts:write': 'Create and update contacts',
'conversations:read': 'List and read conversations',
'broadcasts:send': 'Launch broadcast campaigns',
'webhooks:manage': 'Register and manage outbound event webhooks',
};
/** Type-narrow an unknown value into a valid `ApiScope`. */

135
src/lib/webhooks/deliver.test.ts Normal file
View File

@ -0,0 +1,135 @@
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import type { SupabaseClient } from '@supabase/supabase-js';
vi.mock('@/lib/whatsapp/encryption', () => ({
decrypt: (s: string) => s,
encrypt: (s: string) => s,
}));
// Control the SSRF guard per-test.
vi.mock('@/lib/webhooks/ssrf', () => ({
isDeliverableUrl: vi.fn(async () => true),
}));
import { dispatchWebhookEvent, MAX_CONSECUTIVE_FAILURES } from './deliver';
import { isDeliverableUrl } from './ssrf';
interface Row {
id: string;
url: string;
secret: string;
}
interface Calls {
updates: { id: string; payload: Record<string, unknown> }[];
rpcs: { name: string; args: Record<string, unknown> }[];
}
function makeDb(rows: Row[], calls: Calls) {
const from = () => {
let mode: 'select' | 'update' = 'select';
let payload: Record<string, unknown> = {};
let id: string | null = null;
const b: Record<string, unknown> = {
select: () => b,
eq: (col: string, val: string) => {
if (col === 'id') id = val;
return b;
},
update: (p: Record<string, unknown>) => {
mode = 'update';
payload = p;
return b;
},
contains: () => Promise.resolve({ data: rows, error: null }),
then: (resolve: (v: unknown) => unknown) => {
if (mode === 'update' && id) calls.updates.push({ id, payload });
return resolve({ data: null, error: null });
},
};
return b;
};
const rpc = (name: string, args: Record<string, unknown>) => {
calls.rpcs.push({ name, args });
return Promise.resolve({ data: null, error: null });
};
return { from, rpc } as unknown as SupabaseClient;
}
const emptyCalls = (): Calls => ({ updates: [], rpcs: [] });
beforeEach(() => {
vi.mocked(isDeliverableUrl).mockResolvedValue(true);
vi.stubGlobal('fetch', vi.fn());
});
afterEach(() => vi.unstubAllGlobals());
describe('dispatchWebhookEvent', () => {
it('signs + POSTs (no redirect follow) and resets failure_count on success', async () => {
const fetchMock = vi.fn().mockResolvedValue({ ok: true, status: 200 } as Response);
vi.stubGlobal('fetch', fetchMock);
const calls = emptyCalls();
await dispatchWebhookEvent(
makeDb([{ id: 'a', url: 'https://a.test/hook', secret: 's1' }], calls),
'acct-1',
'message.received',
{ x: 1 }
);
expect(fetchMock).toHaveBeenCalledTimes(1);
const [url, opts] = fetchMock.mock.calls[0];
expect(url).toBe('https://a.test/hook');
expect(opts.redirect).toBe('manual');
expect(opts.headers['X-Wacrm-Event']).toBe('message.received');
expect(opts.headers['X-Wacrm-Signature']).toMatch(/^t=\d+,v1=[0-9a-f]{64}$/);
// Payload carries a dedupe id.
expect(JSON.parse(opts.body).id).toMatch(/[0-9a-f-]{36}/);
expect(calls.updates[0]).toMatchObject({ id: 'a', payload: { failure_count: 0 } });
expect(calls.rpcs).toHaveLength(0);
});
it('records an atomic failure (RPC) when the endpoint errors', async () => {
vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 500 } as Response));
const calls = emptyCalls();
await dispatchWebhookEvent(
makeDb([{ id: 'b', url: 'https://b.test/hook', secret: 's2' }], calls),
'acct-1',
'message.received',
{}
);
expect(calls.rpcs[0]).toEqual({
name: 'record_webhook_failure',
args: { endpoint_id: 'b', max_failures: MAX_CONSECUTIVE_FAILURES },
});
expect(calls.updates).toHaveLength(0);
});
it('blocks a non-public target (SSRF guard) without fetching', async () => {
vi.mocked(isDeliverableUrl).mockResolvedValue(false);
const fetchMock = vi.fn();
vi.stubGlobal('fetch', fetchMock);
const calls = emptyCalls();
await dispatchWebhookEvent(
makeDb([{ id: 'c', url: 'https://127.0.0.1/hook', secret: 's3' }], calls),
'acct-1',
'message.received',
{}
);
expect(fetchMock).not.toHaveBeenCalled();
expect(calls.rpcs[0].name).toBe('record_webhook_failure');
});
it('does nothing when no endpoints are subscribed', async () => {
const fetchMock = vi.fn();
vi.stubGlobal('fetch', fetchMock);
const calls = emptyCalls();
await dispatchWebhookEvent(makeDb([], calls), 'acct-1', 'message.received', {});
expect(fetchMock).not.toHaveBeenCalled();
expect(calls.rpcs).toHaveLength(0);
expect(calls.updates).toHaveLength(0);
});
});

158
src/lib/webhooks/deliver.ts Normal file
View File

@ -0,0 +1,158 @@
// ============================================================
// Outbound webhook delivery.
//
// `dispatchWebhookEvent` finds the account's active endpoints
// subscribed to an event, signs one JSON payload, and POSTs it to
// each in parallel. It is best-effort and never throws — callers fire
// it from the inbound webhook's `after()` block, where a failed
// delivery must not affect the 200 OK returned to Meta.
//
// Delivery semantics (documented in docs/public-api.md):
// - At-most-once per event, single attempt with a short timeout.
// - Each consecutive failure bumps `failure_count`; once it crosses
// MAX_CONSECUTIVE_FAILURES the endpoint is auto-disabled
// (`is_active = false`) so a dead sink stops being hit. A success
// resets the counter and stamps `last_delivery_at`.
// - Durable retry-with-backoff would need a queue/worker (a
// follow-up); in-process retries inside `after()` would burn the
// route's duration budget without a real durability guarantee.
// ============================================================
import { randomUUID } from 'node:crypto';
import type { SupabaseClient } from '@supabase/supabase-js';
import { decrypt } from '@/lib/whatsapp/encryption';
import { buildSignatureHeader } from '@/lib/webhooks/sign';
import { isDeliverableUrl } from '@/lib/webhooks/ssrf';
import type { WebhookEvent } from '@/lib/webhooks/events';
/** Per-endpoint HTTP timeout. Kept short — this runs in `after()`. */
export const DELIVERY_TIMEOUT_MS = 5000;
/** Auto-disable an endpoint after this many consecutive failures. */
export const MAX_CONSECUTIVE_FAILURES = 15;
interface EndpointRow {
id: string;
url: string;
secret: string;
}
/**
* Deliver `event` (+ `data`) to every active endpoint of `accountId`
* subscribed to it. Never throws.
*/
export async function dispatchWebhookEvent(
db: SupabaseClient,
accountId: string,
event: WebhookEvent,
data: unknown
): Promise<void> {
try {
const { data: rows, error } = await db
.from('webhook_endpoints')
.select('id, url, secret')
.eq('account_id', accountId)
.eq('is_active', true)
.contains('events', [event]);
if (error || !rows || rows.length === 0) return;
// Sign the exact bytes we send so a receiver can recompute the
// HMAC over the raw request body. `id` is a per-delivery uuid the
// receiver can dedupe on (deliveries are at-least-once and may
// repeat / arrive out of order).
const payload = JSON.stringify({
id: randomUUID(),
event,
occurred_at: new Date().toISOString(),
account_id: accountId,
data,
});
const tsSeconds = Math.floor(Date.now() / 1000);
await Promise.allSettled(
(rows as EndpointRow[]).map((row) =>
deliverOne(db, row, event, payload, tsSeconds)
)
);
} catch (err) {
// Never let a delivery problem bubble into the webhook response.
console.error('[webhooks] dispatch failed:', err);
}
}
async function deliverOne(
db: SupabaseClient,
row: EndpointRow,
event: WebhookEvent,
payload: string,
tsSeconds: number
): Promise<void> {
// SSRF guard: refuse to POST to a host that resolves to a private /
// loopback / link-local address. Counts as a failure so a
// misconfigured internal URL surfaces and eventually auto-disables.
if (!(await isDeliverableUrl(row.url))) {
console.warn('[webhooks] refusing non-public delivery target for', row.id);
await recordFailure(db, row);
return;
}
let secret: string;
try {
secret = decrypt(row.secret);
} catch (err) {
// A row whose secret can't be decrypted can never produce a valid
// signature — count it as a failure so it eventually auto-disables.
console.error('[webhooks] secret decrypt failed for', row.id, err);
await recordFailure(db, row);
return;
}
try {
const res = await fetch(row.url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Wacrm-Event': event,
'X-Wacrm-Webhook-Id': row.id,
'X-Wacrm-Signature': buildSignatureHeader(payload, secret, tsSeconds),
},
body: payload,
// Do NOT follow redirects — a public URL could 3xx-bounce to an
// internal address, bypassing the SSRF check above. A 3xx is a
// misconfiguration; treat it as a failure.
redirect: 'manual',
signal: AbortSignal.timeout(DELIVERY_TIMEOUT_MS),
});
if (!res.ok) throw new Error(`endpoint responded ${res.status}`);
// Success: clear the failure streak.
await db
.from('webhook_endpoints')
.update({ failure_count: 0, last_delivery_at: new Date().toISOString() })
.eq('id', row.id);
} catch (err) {
console.warn(
`[webhooks] delivery to ${row.id} failed:`,
err instanceof Error ? err.message : err
);
await recordFailure(db, row);
}
}
async function recordFailure(db: SupabaseClient, row: EndpointRow): Promise<void> {
// Atomic increment (+ auto-disable at the threshold) via a SQL
// function — a read-modify-write here would lose increments when two
// deliveries to the same endpoint run concurrently (e.g.
// conversation.created + message.received for one inbound message),
// so a dead endpoint might never reach the disable threshold.
const { error } = await db.rpc('record_webhook_failure', {
endpoint_id: row.id,
max_failures: MAX_CONSECUTIVE_FAILURES,
});
if (error) {
console.error('[webhooks] record_webhook_failure failed for', row.id, error);
}
}

59
src/lib/webhooks/endpoints.test.ts Normal file
View File

@ -0,0 +1,59 @@
import { describe, it, expect } from 'vitest';
import {
generateWebhookSecret,
serializeWebhookEndpoint,
normalizeWebhookUrl,
WEBHOOK_SECRET_PREFIX,
} from './endpoints';
describe('generateWebhookSecret', () => {
it('is prefixed and high-entropy, and unique per call', () => {
const a = generateWebhookSecret();
const b = generateWebhookSecret();
expect(a.startsWith(WEBHOOK_SECRET_PREFIX)).toBe(true);
expect(a.length).toBeGreaterThan(WEBHOOK_SECRET_PREFIX.length + 20);
expect(a).not.toBe(b);
});
});
describe('serializeWebhookEndpoint', () => {
it('projects public fields and never leaks the secret', () => {
const out = serializeWebhookEndpoint({
id: 'w1',
account_id: 'acct',
created_by: 'u1',
url: 'https://example.com/hook',
secret: 'encrypted-blob',
events: ['message.received'],
is_active: true,
last_delivery_at: null,
failure_count: 0,
created_at: '2026-01-01T00:00:00Z',
});
expect(out).not.toHaveProperty('secret');
expect(out).not.toHaveProperty('account_id');
expect(out).toEqual({
id: 'w1',
url: 'https://example.com/hook',
events: ['message.received'],
is_active: true,
last_delivery_at: null,
failure_count: 0,
created_at: '2026-01-01T00:00:00Z',
});
});
});
describe('normalizeWebhookUrl', () => {
it('accepts https and normalizes', () => {
expect(normalizeWebhookUrl(' https://example.com/hook ')).toBe(
'https://example.com/hook'
);
});
it('rejects http, non-URLs, and non-strings', () => {
expect(normalizeWebhookUrl('http://example.com/hook')).toBeNull();
expect(normalizeWebhookUrl('not a url')).toBeNull();
expect(normalizeWebhookUrl(123)).toBeNull();
});
});

66
src/lib/webhooks/endpoints.ts Normal file
View File

@ -0,0 +1,66 @@
// ============================================================
// Webhook endpoint store helpers — secret generation + the public
// (secret-free) serialization used by the management API.
//
// The signing secret is stored AES-256-GCM-encrypted at rest (see
// migration 028) and returned in plaintext exactly once, at creation.
// ============================================================
import { randomBytes } from 'node:crypto';
/** Secret prefix — self-identifying, like `wacrm_live_` for keys. */
export const WEBHOOK_SECRET_PREFIX = 'whsec_';
/**
* Columns safe to return over the API — everything except the
* (encrypted) `secret`, which is only ever surfaced once at creation.
*/
export const WEBHOOK_PUBLIC_COLUMNS =
'id, url, events, is_active, last_delivery_at, failure_count, created_at';
export interface ApiWebhookEndpoint {
id: string;
url: string;
events: string[];
is_active: boolean;
last_delivery_at: string | null;
failure_count: number;
created_at: string;
}
/** Generate a fresh signing secret (full-entropy, URL/header-safe). */
export function generateWebhookSecret(): string {
return `${WEBHOOK_SECRET_PREFIX}${randomBytes(32).toString('base64url')}`;
}
/** Project a `WEBHOOK_PUBLIC_COLUMNS` row into the API shape. */
export function serializeWebhookEndpoint(
row: Record<string, unknown>
): ApiWebhookEndpoint {
return {
id: row.id as string,
url: row.url as string,
events: (row.events as string[] | null) ?? [],
is_active: Boolean(row.is_active),
last_delivery_at: (row.last_delivery_at as string | null) ?? null,
failure_count: (row.failure_count as number | null) ?? 0,
created_at: row.created_at as string,
};
}
/**
* Validate a webhook target URL: must be a well-formed absolute
* `https://` URL (an unencrypted `http://` sink would leak signed
* event payloads). Returns the normalized string or null.
*/
export function normalizeWebhookUrl(input: unknown): string | null {
if (typeof input !== 'string') return null;
const trimmed = input.trim();
try {
const u = new URL(trimmed);
if (u.protocol !== 'https:') return null;
return u.toString();
} catch {
return null;
}
}

40
src/lib/webhooks/events.test.ts Normal file
View File

@ -0,0 +1,40 @@
import { describe, it, expect } from 'vitest';
import {
WEBHOOK_EVENTS,
WEBHOOK_EVENT_DESCRIPTIONS,
isWebhookEvent,
normalizeEvents,
} from './events';
describe('isWebhookEvent', () => {
it('accepts every declared event and rejects others', () => {
for (const e of WEBHOOK_EVENTS) expect(isWebhookEvent(e)).toBe(true);
expect(isWebhookEvent('message.deleted')).toBe(false);
expect(isWebhookEvent(42)).toBe(false);
});
});
describe('every event has a description', () => {
it('covers the vocabulary', () => {
for (const e of WEBHOOK_EVENTS) {
expect(WEBHOOK_EVENT_DESCRIPTIONS[e]).toBeTruthy();
}
});
});
describe('normalizeEvents', () => {
it('de-duplicates a valid list', () => {
expect(
normalizeEvents(['message.received', 'message.received', 'conversation.created'])
).toEqual(['message.received', 'conversation.created']);
});
it('rejects an unknown event', () => {
expect(normalizeEvents(['message.received', 'nope'])).toBeNull();
});
it('rejects a non-array and an empty array', () => {
expect(normalizeEvents('message.received')).toBeNull();
expect(normalizeEvents([])).toBeNull();
});
});

48
src/lib/webhooks/events.ts Normal file
View File

@ -0,0 +1,48 @@
// ============================================================
// Outbound webhook event vocabulary — pure, no I/O.
//
// An endpoint subscribes to one or more of these. Adding an event is
// one entry here plus a `dispatchWebhookEvent` call at the source of
// the event (the DB stores subscriptions as a free `text[]`, so no
// migration is needed — same model as API scopes).
// ============================================================
export const WEBHOOK_EVENTS = [
'message.received', // an inbound WhatsApp message landed
'message.status_updated', // a sent message advanced (sent/delivered/read)
'conversation.created', // a new conversation was opened for a contact
] as const;
export type WebhookEvent = (typeof WEBHOOK_EVENTS)[number];
/** Human-readable descriptions (surfaced in docs / a future UI). */
export const WEBHOOK_EVENT_DESCRIPTIONS: Record<WebhookEvent, string> = {
'message.received': 'An inbound message was received from a contact',
'message.status_updated':
'A message you sent changed delivery status (sent/delivered/read/failed)',
'conversation.created': 'A new conversation was opened',
};
/** Type-narrow an unknown value into a valid `WebhookEvent`. */
export function isWebhookEvent(value: unknown): value is WebhookEvent {
return (
typeof value === 'string' &&
(WEBHOOK_EVENTS as readonly string[]).includes(value)
);
}
/**
* Validate + de-duplicate a caller-supplied event list. Returns the
* cleaned list, or `null` if any entry is unknown (callers turn that
* into a 400). An empty list is rejected as `null` too — an endpoint
* subscribed to nothing is almost certainly a mistake.
*/
export function normalizeEvents(input: unknown): WebhookEvent[] | null {
if (!Array.isArray(input) || input.length === 0) return null;
const out: WebhookEvent[] = [];
for (const entry of input) {
if (!isWebhookEvent(entry)) return null;
if (!out.includes(entry)) out.push(entry);
}
return out;
}

45
src/lib/webhooks/sign.test.ts Normal file
View File

@ -0,0 +1,45 @@
import { describe, it, expect } from 'vitest';
import { buildSignatureHeader, verifySignatureHeader } from './sign';
const secret = 'whsec_testsecret';
const body = JSON.stringify({ event: 'message.received', data: { a: 1 } });
describe('buildSignatureHeader', () => {
it('emits the t=,v1= shape and is deterministic', () => {
const h1 = buildSignatureHeader(body, secret, 1_700_000_000);
const h2 = buildSignatureHeader(body, secret, 1_700_000_000);
expect(h1).toMatch(/^t=1700000000,v1=[0-9a-f]{64}$/);
expect(h1).toBe(h2);
});
});
describe('verifySignatureHeader', () => {
const now = 1_700_000_000;
const header = buildSignatureHeader(body, secret, now);
it('accepts a valid, in-tolerance signature', () => {
expect(verifySignatureHeader(header, body, secret, now + 10)).toBe(true);
});
it('rejects a tampered body', () => {
expect(verifySignatureHeader(header, body + 'x', secret, now)).toBe(false);
});
it('rejects a wrong secret', () => {
expect(verifySignatureHeader(header, body, 'whsec_other', now)).toBe(false);
});
it('rejects a stale timestamp (replay protection)', () => {
expect(verifySignatureHeader(header, body, secret, now + 10_000)).toBe(false);
});
it('rejects a malformed header', () => {
expect(verifySignatureHeader('garbage', body, secret, now)).toBe(false);
});
it('tolerates uppercase hex and whitespace in the header', () => {
const [, t, v1] = header.match(/^t=(\d+),v1=([0-9a-f]+)$/)!;
const loose = `t=${t}, v1=${v1.toUpperCase()}`;
expect(verifySignatureHeader(loose, body, secret, Number(t))).toBe(true);
});
});

68
src/lib/webhooks/sign.ts Normal file
View File

@ -0,0 +1,68 @@
// ============================================================
// Webhook payload signing — pure, server-side.
//
// Every delivery carries an `X-Wacrm-Signature` header so receivers
// can verify the request really came from wacrm and wasn't tampered
// with or replayed. The scheme is Stripe-style:
//
// X-Wacrm-Signature: t=<unix_seconds>,v1=<hex HMAC-SHA256>
//
// where the signed message is `${t}.${rawBody}` and the key is the
// endpoint's secret. Receivers recompute the HMAC over the raw body
// they received (not a re-serialized copy) and compare in constant
// time, and reject if `t` is too old (replay protection).
// ============================================================
import { createHmac, timingSafeEqual } from 'node:crypto';
/**
* Build the `X-Wacrm-Signature` header value for `rawBody`, signed
* with `secret` at time `timestampSeconds` (pass it in — never call
* Date.now() here, so the value is testable and callers control the
* clock).
*/
export function buildSignatureHeader(
rawBody: string,
secret: string,
timestampSeconds: number
): string {
const signature = createHmac('sha256', secret)
.update(`${timestampSeconds}.${rawBody}`)
.digest('hex');
return `t=${timestampSeconds},v1=${signature}`;
}
/**
* Verify a signature header. Exposed so a wacrm-to-wacrm integration
* (or a test) can validate deliveries; receivers in other stacks
* reimplement the same three lines. `toleranceSeconds` bounds replay.
*/
export function verifySignatureHeader(
header: string,
rawBody: string,
secret: string,
nowSeconds: number,
toleranceSeconds = 300
): boolean {
const parts = Object.fromEntries(
header.split(',').map((kv) => {
const i = kv.indexOf('=');
return [kv.slice(0, i).trim(), kv.slice(i + 1)];
})
);
const t = Number(parts.t);
// Normalize the presented signature: hex is case-insensitive and a
// header may carry stray whitespace (e.g. `t=…, v1=…`), so lower-case
// and trim before the constant-time compare.
const v1 = typeof parts.v1 === 'string' ? parts.v1.trim().toLowerCase() : '';
if (!Number.isFinite(t) || !v1) return false;
if (Math.abs(nowSeconds - t) > toleranceSeconds) return false;
const expected = createHmac('sha256', secret)
.update(`${t}.${rawBody}`)
.digest('hex');
// Constant-time compare; guard against length mismatch (timingSafeEqual
// throws on unequal-length buffers).
if (expected.length !== v1.length) return false;
return timingSafeEqual(Buffer.from(expected), Buffer.from(v1));
}

50
src/lib/webhooks/ssrf.test.ts Normal file
View File

@ -0,0 +1,50 @@
import { describe, it, expect } from 'vitest';
import { isPrivateOrReservedIp, isDeliverableUrl } from './ssrf';
describe('isPrivateOrReservedIp', () => {
it('flags loopback / private / link-local / CGNAT IPv4', () => {
for (const ip of [
'127.0.0.1',
'10.0.0.5',
'172.16.0.1',
'172.31.255.255',
'192.168.1.1',
'169.254.169.254', // cloud metadata
'100.64.0.1', // CGNAT
'0.0.0.0',
]) {
expect(isPrivateOrReservedIp(ip)).toBe(true);
}
});
it('allows public IPv4', () => {
for (const ip of ['8.8.8.8', '1.1.1.1', '172.15.0.1', '172.32.0.1', '93.184.216.34']) {
expect(isPrivateOrReservedIp(ip)).toBe(false);
}
});
it('flags loopback / ULA / link-local IPv6 and IPv4-mapped privates', () => {
for (const ip of ['::1', 'fe80::1', 'fc00::1', 'fd12::34', '::ffff:127.0.0.1']) {
expect(isPrivateOrReservedIp(ip)).toBe(true);
}
expect(isPrivateOrReservedIp('2606:4700:4700::1111')).toBe(false);
});
});
describe('isDeliverableUrl', () => {
it('rejects literal private IPs and internal names without DNS', async () => {
expect(await isDeliverableUrl('https://127.0.0.1/hook')).toBe(false);
expect(await isDeliverableUrl('https://169.254.169.254/latest/meta-data')).toBe(false);
expect(await isDeliverableUrl('https://[::1]/hook')).toBe(false);
expect(await isDeliverableUrl('https://localhost/hook')).toBe(false);
expect(await isDeliverableUrl('https://foo.internal/hook')).toBe(false);
});
it('rejects a malformed URL', async () => {
expect(await isDeliverableUrl('not a url')).toBe(false);
});
it('allows a literal public IP', async () => {
expect(await isDeliverableUrl('https://8.8.8.8/hook')).toBe(true);
});
});

83
src/lib/webhooks/ssrf.ts Normal file
View File

@ -0,0 +1,83 @@
// ============================================================
// SSRF guard for outbound webhook delivery.
//
// A webhook URL is attacker-influenced (any account admin with
// `webhooks:manage` can register one) and our server makes the request,
// so an unguarded fetch is a Server-Side Request Forgery primitive: a
// URL pointing at `127.0.0.1`, a cloud metadata IP (`169.254.169.254`),
// or an RFC1918 host would let a caller probe / POST to internal
// services from the app's network.
//
// `isDeliverableUrl` resolves the host and rejects any address that is
// loopback, private, link-local, ULA, or otherwise non-publicly-
// routable. Combined with `redirect: 'manual'` at the call site (so a
// public URL can't 3xx-bounce to an internal one), this blocks the
// common SSRF vectors. It is NOT a defense against DNS rebinding (a
// host that resolves public here but flips to private before connect) —
// that needs pinning the resolved IP into the socket, which fetch
// doesn't expose; documented as a residual risk.
// ============================================================
import { lookup } from 'node:dns/promises';
import { isIP } from 'node:net';
/** True for loopback / private / link-local / reserved IPv4 or IPv6. */
export function isPrivateOrReservedIp(ip: string): boolean {
const v4 = ip.match(/^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/);
if (v4) {
const a = Number(v4[1]);
const b = Number(v4[2]);
if (a === 0) return true; // "this" network
if (a === 10) return true; // private
if (a === 127) return true; // loopback
if (a === 169 && b === 254) return true; // link-local + cloud metadata
if (a === 172 && b >= 16 && b <= 31) return true; // private
if (a === 192 && b === 168) return true; // private
if (a === 100 && b >= 64 && b <= 127) return true; // CGNAT
return false;
}
const v6 = ip.toLowerCase().replace(/^\[|\]$/g, '');
if (v6 === '::1' || v6 === '::') return true; // loopback / unspecified
if (v6.startsWith('fe8') || v6.startsWith('fe9') || v6.startsWith('fea') || v6.startsWith('feb'))
return true; // fe80::/10 link-local
if (v6.startsWith('fc') || v6.startsWith('fd')) return true; // fc00::/7 ULA
const mapped = v6.match(/::ffff:(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$/);
if (mapped) return isPrivateOrReservedIp(mapped[1]); // IPv4-mapped
return false;
}
/**
* True if `rawUrl`'s host resolves only to publicly-routable
* address(es). Returns false for a malformed URL, an obvious internal
* name (`localhost`, `*.local`, `*.internal`), a literal private IP, or
* a hostname that resolves to any private/reserved address.
*/
export async function isDeliverableUrl(rawUrl: string): Promise<boolean> {
let host: string;
try {
host = new URL(rawUrl).hostname.replace(/^\[|\]$/g, '');
} catch {
return false;
}
if (isIP(host)) return !isPrivateOrReservedIp(host);
const lower = host.toLowerCase();
if (
lower === 'localhost' ||
lower.endsWith('.localhost') ||
lower.endsWith('.local') ||
lower.endsWith('.internal')
) {
return false;
}
try {
const results = await lookup(host, { all: true });
if (results.length === 0) return false;
return results.every((r) => !isPrivateOrReservedIp(r.address));
} catch {
return false; // unresolvable → not deliverable
}
}

103
supabase/migrations/028_webhook_endpoints.sql Normal file
View File

@ -0,0 +1,103 @@
-- ============================================================
-- 028_webhook_endpoints.sql — Outbound event webhooks (public API)
--
-- Lets an account register HTTPS endpoints that wacrm POSTs to when
-- something happens (an inbound message arrives, a delivery status
-- changes, a conversation is created). This is the "react to inbound"
-- half of the public API (#245): instead of polling
-- `GET /api/v1/conversations`, an automation subscribes once and is
-- pushed the events it cares about.
--
-- Design notes
-- - Account-scoped, never user-scoped (same as `api_keys`).
-- `created_by` records who registered it (audit); ON DELETE SET
-- NULL so removing a teammate doesn't drop their integration's
-- endpoint.
-- - `secret` is the HMAC signing key. UNLIKE `api_keys` (where we
-- store only a hash because the key is a bearer credential the
-- *client* presents), here *we* sign each outgoing payload with
-- the secret and the receiver verifies it — so we need the
-- plaintext at delivery time. We store it AES-256-GCM-encrypted
-- at rest (same `encrypt()`/`decrypt()` as `whatsapp_config.
-- access_token`), and return the plaintext to the creator exactly
-- once so they can configure their verifier.
-- - `events[]` is the subscription filter (free text[], validated
-- in the app layer against `src/lib/webhooks/events.ts` — a new
-- event type is a code change, not a migration, mirroring scopes).
-- - `failure_count` counts *consecutive* delivery failures; the
-- deliverer auto-sets `is_active = false` once it crosses a
-- threshold so a permanently-dead endpoint stops being retried.
-- A successful delivery resets it to 0.
--
-- RLS
-- Settings-class, mirroring `api_keys`: any member may read the
-- roster; only admin+ may create/update/delete. The delivery path
-- and the public-API management routes both use the service-role
-- client (an API caller has no `auth.uid()`), so RLS is the guard
-- for any dashboard UI that reads the table directly.
--
-- Idempotent — safe to run multiple times.
-- ============================================================
CREATE TABLE IF NOT EXISTS webhook_endpoints (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
account_id uuid NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
created_by uuid REFERENCES auth.users(id) ON DELETE SET NULL,
url text NOT NULL, -- HTTPS endpoint we POST to
secret text NOT NULL, -- AES-256-GCM-encrypted HMAC signing secret
events text[] NOT NULL DEFAULT '{}',
is_active boolean NOT NULL DEFAULT true,
last_delivery_at timestamptz, -- last successful delivery
failure_count integer NOT NULL DEFAULT 0, -- consecutive failures; reset to 0 on success
created_at timestamptz NOT NULL DEFAULT now()
);
-- Every delivery + management query filters by account_id.
CREATE INDEX IF NOT EXISTS webhook_endpoints_account_id_idx
ON webhook_endpoints (account_id);
ALTER TABLE webhook_endpoints ENABLE ROW LEVEL SECURITY;
-- SELECT: any member of the account (viewer+) can see the roster.
DROP POLICY IF EXISTS webhook_endpoints_select ON webhook_endpoints;
CREATE POLICY webhook_endpoints_select ON webhook_endpoints FOR SELECT
USING (is_account_member(account_id));
-- INSERT / UPDATE / DELETE: admin+ only (settings-class).
DROP POLICY IF EXISTS webhook_endpoints_insert ON webhook_endpoints;
CREATE POLICY webhook_endpoints_insert ON webhook_endpoints FOR INSERT
WITH CHECK (is_account_member(account_id, 'admin'));
DROP POLICY IF EXISTS webhook_endpoints_update ON webhook_endpoints;
CREATE POLICY webhook_endpoints_update ON webhook_endpoints FOR UPDATE
USING (is_account_member(account_id, 'admin'));
DROP POLICY IF EXISTS webhook_endpoints_delete ON webhook_endpoints;
CREATE POLICY webhook_endpoints_delete ON webhook_endpoints FOR DELETE
USING (is_account_member(account_id, 'admin'));
-- ============================================================
-- Atomic consecutive-failure counter.
--
-- The deliverer records failures through this function rather than a
-- read-modify-write: two deliveries to the same endpoint can run
-- concurrently (e.g. conversation.created + message.received for one
-- inbound message), and a client-side `count = count + 1` would lose
-- increments, so a dead endpoint might never reach the auto-disable
-- threshold. The `+ 1` and the disable decision happen in one UPDATE.
-- Only ever disables (never re-enables) — re-enabling is an explicit
-- PATCH by an admin, which resets the counter.
-- ============================================================
CREATE OR REPLACE FUNCTION public.record_webhook_failure(
endpoint_id uuid,
max_failures int
)
RETURNS void AS $$
UPDATE webhook_endpoints
SET failure_count = failure_count + 1,
is_active = CASE
WHEN failure_count + 1 >= max_failures THEN false
ELSE is_active
END
WHERE id = endpoint_id;
$$ LANGUAGE sql SECURITY DEFINER SET search_path = public;