sax3l/accounted
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(api): Phase 6 PR-3 — substrate hardening (SKIP LOCKED + DNS pinning + test debt) (#500)

Browse Source 
* feat(api): operations table immutability trigger

BFNAR 2013:2 kap 8 § behandlingshistorik integrity: once an operations
row is in a terminal status (succeeded / failed / cancelled) the audit
record of what happened becomes immutable. Adds the BEFORE UPDATE and
BEFORE DELETE triggers that the webhook_deliveries table already has
(20260515170000 / 20260515190000), mirroring their predicate shape and
error code exactly.

Closes the Phase 4 PR-2 (PR #469) review-round carry-over flagged by
Swedish-compliance: previously a future bug, a privileged operator, or
a compromised service-role caller could rewrite "this year-end close
succeeded" to "failed" by updating an already-terminal row. The
running → succeeded/failed/cancelled transition itself stays legal
because the trigger keys on OLD.status, which is non-terminal at the
moment of the legitimate UPDATE.

pg test covers all transitions (allowed and blocked) plus DELETE on
both terminal and non-terminal rows.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(api): atomic SKIP LOCKED claim for webhook dispatch

Replaces the SELECT-then-UPDATE-intersect pattern in the dispatcher
with a single-roundtrip SQL function using FOR UPDATE SKIP LOCKED.
PostgREST can't express SKIP LOCKED through the JS client, so the
previous shape relied on a CAS guard inside an UPDATE WHERE status IN
('pending','failed') to ensure only one of two overlapping cron ticks
claimed any given row.

The CAS pattern was correct (under load — receivers >60s could push a
batch past the next minute's tick) but burned two round trips and
forced the application to negotiate the locking semantics in JS.
The function form moves the contention to the DB, where SKIP LOCKED
makes a row held by a concurrent tick simply invisible to the second
caller. One round trip, no JS-side intersect.

All filter semantics are preserved verbatim inside the function:
status IN ('pending','failed'), next_attempt_at <= now, webhook_id
IS NOT NULL, ORDER BY next_attempt_at ASC, LIMIT batchSize. p_batch_size
is bounded (0, 1000] to forestall a runaway lock-set in case a caller
misconfigures it.

pg test covers basic claim (pending + failed), future-due skip, dangling-
row (webhook_id IS NULL) skip, terminal-status skip, batch-size limits,
out-of-range argument rejection, and the SKIP LOCKED invariant itself
using two concurrent pool clients in BEGIN — the second caller does not
see the row A locked, no double-delivery.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(api): pinned-IP HTTPS dispatch (close DNS-rebinding window)

The url-guard.ts file header openly flagged the remaining gap:
"a separate DNS-rebinding window (between dispatch-time validation
and the actual fetch) remains; closing that requires a custom HTTPS
agent that pins the resolved IP — tracked for follow-up." This closes it.

The previous shape was:
  1. validateWebhookUrl()  → DNS resolves to [public IP], returns ok
  2. fetch(webhook_url)    → re-resolves DNS; an attacker who flipped
                             the A record in the interval gets a
                             private-IP socket

The new pinnedHttpsFetch helper validates DNS once, then opens a
node:https.request to that pinned IP — but keeps the original hostname
in the TLS SNI extension (so the receiver's cert validates) and in the
HTTP Host header (so vhost routing still works). The request socket
never re-resolves DNS, foreclosing the rebind race entirely.

Built on node:https.request rather than undici's Agent so the project
doesn't take on a new dep — the stdlib API is also more explicit about
the SNI / Host / pinned-IP split. Test seam injects both validateUrl
and httpsRequest so the unit tests verify the pinning shape without
standing up an HTTPS server.

The dispatcher's attemptDelivery is rewritten as a switch over the four
PinnedFetchResult kinds (ok / unsafe_url / redirect_blocked / timeout
/ transport_error). The previous fetch-based code path that distinguished
redirect rejection by string-matching err.message is gone — the new
result type makes the distinction structural.

8 unit tests cover the SNI/Host/pinned-IP shape, port handling,
redirect_blocked, transport_error, timeout, response-body truncation,
first-IP determinism, and the validation short-circuit (never opens a
socket when the URL fails the SSRF guard).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(api): pg tests for webhook substrate triggers (PR-1 test debt)

CLAUDE.md ("Testing" + "Migration Rules") mandates a *.pg.test.ts for
any PR touching a trigger / RPC / RLS / DEFERRABLE constraint. Phase 6
PR-1 (#496) shipped three webhook_deliveries triggers without the
accompanying pg test; this closes that debt.

Triggers covered:
  - enforce_webhook_delivery_immutability  (BEFORE UPDATE)
  - block_webhook_delivery_terminal_delete (BEFORE DELETE)
  - assert_webhook_delivery_company_match  (BEFORE INSERT)

13 cases verify the lifecycle the dispatcher depends on remains mutable
(pending → in_flight, in_flight → failed, failed → in_flight, in_flight
→ delivered) while terminal-status rows (delivered / dead) are write-
locked and the cross-tenant INSERT path is refused with the
ERRCODE=check_violation contract documented in the migration.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(api): integration tests for webhook routes (PR-1 test debt)

CLAUDE.md mandates integration tests under app/api/v1/ for every route.
Phase 6 PR-1 (#496) shipped the eight v1 webhook routes (five under
/companies/{companyId}/webhooks/ + the cross-tenant /webhook-deliveries/
{id}/retry) without them; closes that debt.

19 cases for the /webhooks/ verticals:
  POST     /webhooks                    create + secret-once + payroll-scope gate + SSRF
  GET      /webhooks                    list (no secret) + empty list
  GET      /webhooks/:id                detail (no secret) + 404
  PATCH    /webhooks/:id                update + active=true re-enable + SSRF re-check + empty-body
  DELETE   /webhooks/:id                204 hard delete
  POST     /webhooks/:id/test           enqueue + 404 + disabled-rejection
  GET      /webhooks/:id/deliveries     happy path + ownership 404

7 cases for the retry route:
  POST /webhook-deliveries/:id/retry   dead → fresh pending row, live-status refusal,
                                       cross-tenant 404, disabled-webhook gate,
                                       SSRF re-check, delivery 404, webhook-gone 404

Both files mirror the suppliers/customers integration test pattern:
Proxy-backed Supabase mock with per-table queues, validateApiKey +
validateWebhookUrl stubbed to control auth and DNS deterministically.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(api): address PR-500 review round 1 — pg-real CI fix + 4 review items

1. pg-real CI was red on this PR: the new webhook trigger pg.test.ts and
   claim-due-webhook-deliveries pg.test.ts fixtures tried to INSERT into
   `webhooks.user_id`, which doesn't exist in the migration history. The
   column was never declared in automation_webhooks (20260415000000) nor
   added by webhooks_v2 (20260515170000) — so a fresh schema replay had
   no such column. The webhook create route (`webhooks.create`) was also
   referencing this non-existent column in its INSERT, so the production
   route was latent-broken since PR-1 and never exercised against a fresh
   DB. Drop the `user_id` field from both the route INSERT and the pg
   fixtures. Actor attribution lives on `created_by_api_key_id` (which
   leads back to the owning user via `api_keys.user_id`).

2. Greptile P2 #1 — `recoverStuckInFlight` carried a redundant
   `.not('status','in','(delivered,dead)')` filter alongside
   `.eq('status','in_flight')`, with a comment that incorrectly described
   PostgreSQL's UPDATE re-evaluation semantics. Under READ COMMITTED,
   UPDATE re-evaluates WHERE against each row's CURRENT value when it
   acquires the row lock — a row that raced to terminal status will fail
   `status='in_flight'` on re-evaluation and be skipped, no immutability
   trigger fires. Drop the redundant filter and rewrite the comment.

3. Greptile P2 #2 — added explicit pg test verifying `in_flight` rows are
   skipped by `claim_due_webhook_deliveries`. The status filter is what
   prevents double-delivery and is the entire point of the SKIP LOCKED
   substrate; making that invariant load-bearing in the test suite
   forecloses a future filter expansion silently regressing it.

4. Greptile P2 #3 — pinned-fetch registered both `res.on('end', finalize)`
   and `res.on('close', finalize)`. Node fires BOTH on normal completions,
   so finalize ran twice; the outer `settled` guard squashed the
   double-resolve but the header reconstruction still ran twice. Switch
   to `once` + self-removing pair so finalize runs exactly once on
   whichever event fires first (normal: end; truncation: close).

5. Compliance Swarm V8.2.1 — the retry route only checked
   `webhooks:manage` even when retrying `salary_run.* / agi.*` deliveries.
   Mirror the create-route elevated-scope gate so a key with only
   `webhooks:manage` cannot re-emit payroll payloads carrying
   personnummer / lönesummor / skatteavdrag. New integration test verifies
   the gate returns 403 INSUFFICIENT_SCOPE with `required_scope:
   payroll:read`.

35 tests pass locally (+1 vs pre-fix). Type-check clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(api): address PR-500 review round 2 — 2 small precision fixes

1. Compliance Swarm Art.32 / A.8.24 — response_body size cap was enforced
   only at the application layer (pinnedHttpsFetch's maxResponseBytes=4096
   constant). A future refactor that bypassed the truncation, or a non-
   dispatcher write path into webhook_deliveries.response_body, would
   silently land large blobs in a column adjacent to event payloads
   carrying personal data. Add a CHECK constraint at the DB layer with
   a generous ceiling (8 KB — double the application cap so legitimate
   dispatcher writes never hit it; only a regression surfaces as a
   check_violation).

2. Compliance Swarm CC6.6 — pinned-fetch substitutes the validated IP
   for `host` while keeping the original hostname in `servername`. A
   reader could reasonably worry that the IP substitution weakens TLS
   hostname verification. Document explicitly that Node's default
   `checkServerIdentity` matches the cert's SAN/CN against `servername`
   (not `host`), so a forged endpoint at the pinned IP with a valid
   cert for a different hostname would fail the handshake. No code
   change — the default behavior is correct; the comment forecloses
   future "this looks dangerous" review-round noise on the same line.

Items NOT addressed (with rationale documented elsewhere):

- Compliance Swarm V8.2.1 (retry route 404-vs-404 information leak):
  delivery IDs are UUIDs; the "leak" is the ability to probe existence
  of an opaque 128-bit identifier the caller already has, which is not
  meaningfully different from probing for any opaque token. Both
  branches return the same structured 404 envelope.

- Compliance Swarm CC7.2 (restore the .not() defense-in-depth filter):
  direct contradiction of last round's Greptile P2 fix. Greptile's
  PG-semantics analysis is correct — under READ COMMITTED, UPDATE
  re-evaluates WHERE against the row's current value when it acquires
  the lock, so .eq('status','in_flight') already handles the race.
  Adding a redundant .not() restores a misleading comment without
  closing a real gap. This is the documented Compliance Swarm
  oscillation pattern from the project's Phase 4 lessons.

- Compliance Swarm CC6.1 (webhook secret encryption-at-rest):
  architectural choice from PR-1; not in PR-3 (substrate hardening)
  scope. Belongs to a future hardening PR.

- Swedish-compliance review (operations queued/running rows hard-
  deletable): deliberate operability tradeoff — operators need to
  clear stuck/queued entries that crashed mid-flight. Blocking all
  deletes would force a manual DB intervention every time a worker
  crashed before reaching terminal status. The audit trail starts
  at terminal-state mutation, which IS blocked.

- Swedish-compliance review (salary_run.* / agi.* payload anonymisation
  after 7 years): already on the deferred-list as part of the 90-day
  TTL cleanup cron item from the PR description. Belongs to a
  retention-policy follow-up PR.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jakob Wennberg
2026-05-15 20:38:19 +02:00
committed by GitHub
parent 240412fd32
commit afb21ea638
13 changed files with 2601 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/v1/companies/[companyId]/webhooks/__tests__/route.test.ts
+665
View File
@@ -0,0 +1,665 @@
/**
* Integration tests for the v1 webhooks vertical (Phase 6 PR-1).
*
* Phase 6 PR-1 (#496) shipped the substrate with deferred integration
* tests. This file closes the test debt for the company-scoped routes:
*
* POST /webhooks (create + secret-once)
* GET /webhooks (list, no secret)
* GET /webhooks/:id (detail)
* PATCH /webhooks/:id (update + SSRF re-check)
* DELETE /webhooks/:id (hard delete, audit trail survives)
* POST /webhooks/:id/test (synthetic delivery)
* GET /webhooks/:id/deliveries (delivery audit list)
*
* Retry (POST /webhook-deliveries/:id/retry) is covered in its sibling
* test file under app/api/v1/webhook-deliveries/.
*
* Mirrors the suppliers vertical test pattern: a Proxy-backed Supabase
* mock returns whatever the route awaits, keyed by table name. Focus is
* on outcome (status / body shape) rather than query mechanics — the
* wrapper already validates auth, scope, idempotency, and company
* membership.
*/
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
beforeAll(() => {
if (process.env.NODE_ENV !== 'test') {
throw new Error(
`webhook route tests require NODE_ENV=test (got ${process.env.NODE_ENV ?? 'undefined'})`,
)
}
process.env.NEXT_PUBLIC_SUPABASE_URL ||= 'http://localhost:54321'
process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY ||= 'test-anon-key'
})
vi.mock('@/lib/auth/api-keys', async () => {
const actual = await vi.importActual<typeof import('@/lib/auth/api-keys')>('@/lib/auth/api-keys')
return {
...actual,
validateApiKey: vi.fn(),
createServiceClientNoCookies: vi.fn(),
}
})
vi.mock('@supabase/supabase-js', async () => {
const actual = await vi.importActual<typeof import('@supabase/supabase-js')>('@supabase/supabase-js')
return { ...actual, createClient: vi.fn().mockReturnValue({}) }
})
// SSRF DNS validation lives behind a network call (dns.resolve4/6). Stub
// it so we can deterministically force ok-vs-rejected outcomes; otherwise
// the test would actually resolve example.com and have flaky behavior in
// air-gapped CI.
vi.mock('@/lib/webhooks/url-guard', async () => {
const actual = await vi.importActual<typeof import('@/lib/webhooks/url-guard')>(
'@/lib/webhooks/url-guard',
)
return {
...actual,
validateWebhookUrl: vi.fn(),
}
})
import { validateApiKey, createServiceClientNoCookies } from '@/lib/auth/api-keys'
import { validateWebhookUrl } from '@/lib/webhooks/url-guard'
import { GET as listWebhooks, POST as createWebhook } from '../route'
import {
GET as getWebhook,
PATCH as updateWebhook,
DELETE as deleteWebhook,
} from '../[id]/route'
import { POST as testWebhook } from '../[id]/test/route'
import { GET as listDeliveries } from '../[id]/deliveries/route'
const mockValidate = validateApiKey as ReturnType<typeof vi.fn>
const mockServiceClient = createServiceClientNoCookies as ReturnType<typeof vi.fn>
const mockUrlGuard = validateWebhookUrl as ReturnType<typeof vi.fn>
interface TableResp {
data?: unknown
error?: unknown
count?: number | null
}
function makeFlexibleSupabase(byTable: Record<string, TableResp | TableResp[]>) {
const queues = new Map<string, TableResp[]>()
for (const [t, val] of Object.entries(byTable)) {
queues.set(t, Array.isArray(val) ? [...val] : [val])
}
const buildChain = (table: string): unknown => {
const handler: ProxyHandler<object> = {
get(_target, prop) {
if (prop === 'then') {
return (resolve: (v: unknown) => void) => {
const q = queues.get(table)
const next = q && q.length > 1 ? q.shift()! : (q?.[0] ?? { data: null, error: null })
resolve(next)
}
}
return (..._args: unknown[]) => buildChain(table)
},
}
return new Proxy({}, handler)
}
return { from: vi.fn((table: string) => buildChain(table)) }
}
const COMPANY_ID = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'
const WEBHOOK_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'
const DELIVERY_ID = 'cccccccc-cccc-4ccc-8ccc-cccccccccccc'
const USER_ID = 'user-1'
function makeRequest(url: string, init?: RequestInit): Request {
return new Request(url, {
...init,
headers: {
Authorization: 'Bearer test-fixture-not-a-real-key',
'Idempotency-Key': 'b1aaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa',
...(init?.headers ?? {}),
},
})
}
function companyParams(companyId: string) {
return { params: Promise.resolve({ companyId }) }
}
function detailParams(companyId: string, id: string) {
return { params: Promise.resolve({ companyId, id }) }
}
const SAMPLE_WEBHOOK = {
id: WEBHOOK_ID,
name: 'CRM sync',
description: null,
event_type: 'invoice.paid',
webhook_url: 'https://example.com/hooks/gnubok',
active: true,
api_version_pinned: '2026-05-12',
disabled_at: null,
disabled_reason: null,
created_at: '2026-05-15T12:00:00Z',
updated_at: '2026-05-15T12:00:00Z',
}
beforeEach(() => {
vi.clearAllMocks()
mockValidate.mockResolvedValue({
userId: USER_ID,
companyId: COMPANY_ID,
apiKeyId: 'ak_1',
apiKeyName: 'CI key',
scopes: ['webhooks:manage', 'payroll:read'],
mode: 'live',
})
// Default URL validation: always ok. Tests override per-case.
mockUrlGuard.mockResolvedValue({
ok: true,
hostname: 'example.com',
resolvedAddresses: ['203.0.113.42'],
})
})
// ──────────────────────────────────────────────────────────────────────
// POST /webhooks (create)
// ──────────────────────────────────────────────────────────────────────
describe('POST /api/v1/companies/:companyId/webhooks', () => {
it('returns 201 with the freshly-minted secret on success', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: SAMPLE_WEBHOOK, error: null },
}),
)
const res = await createWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
event_type: 'invoice.paid',
webhook_url: 'https://example.com/hooks/gnubok',
name: 'CRM sync',
}),
}),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(201)
const body = await res.json()
expect(body.data.id).toBe(WEBHOOK_ID)
// Secret is returned EXACTLY ONCE on create. We don't pin the prefix
// shape too tightly — the contract is "non-empty string with whsec_
// prefix" and the schema documents the exact length elsewhere.
expect(typeof body.data.secret).toBe('string')
expect(body.data.secret).toMatch(/^whsec_/)
})
it('returns 400 VALIDATION_ERROR when webhook_url is not https', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
}),
)
const res = await createWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
event_type: 'invoice.paid',
webhook_url: 'http://example.com/hooks',
name: 'CRM sync',
}),
}),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
})
it('returns 400 VALIDATION_ERROR when the SSRF guard rejects the URL', async () => {
mockUrlGuard.mockResolvedValueOnce({
ok: false,
reason: 'private_address',
detail: '10.0.0.1 is private',
})
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
}),
)
const res = await createWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
event_type: 'invoice.paid',
webhook_url: 'https://internal.example/hooks',
name: 'internal',
}),
}),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
expect(body.error.details.reason).toBe('private_address')
})
it('requires payroll:read for salary_run.* event types (elevated-scope gate)', async () => {
// Key has webhooks:manage but NOT payroll:read.
mockValidate.mockResolvedValueOnce({
userId: USER_ID,
companyId: COMPANY_ID,
apiKeyId: 'ak_1',
apiKeyName: 'CI key',
scopes: ['webhooks:manage'],
mode: 'live',
})
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
}),
)
const res = await createWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
event_type: 'salary_run.booked',
webhook_url: 'https://example.com/hooks',
name: 'payroll',
}),
}),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(403)
const body = await res.json()
expect(body.error.code).toBe('INSUFFICIENT_SCOPE')
expect(body.error.details.required_scope).toBe('payroll:read')
})
it('returns 401 UNAUTHORIZED when no Bearer token is supplied', async () => {
const req = new Request(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
event_type: 'invoice.paid',
webhook_url: 'https://example.com/hooks',
name: 'CRM',
}),
})
const res = await createWebhook(req, companyParams(COMPANY_ID))
expect(res.status).toBe(401)
})
})
// ──────────────────────────────────────────────────────────────────────
// GET /webhooks (list)
// ──────────────────────────────────────────────────────────────────────
describe('GET /api/v1/companies/:companyId/webhooks', () => {
it('lists webhooks for the company without exposing secrets', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: [SAMPLE_WEBHOOK], error: null },
}),
)
const res = await listWebhooks(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data.webhooks).toHaveLength(1)
expect(body.data.webhooks[0].id).toBe(WEBHOOK_ID)
// Secret MUST never be in a list response — surfaced only on create.
expect(body.data.webhooks[0]).not.toHaveProperty('secret')
})
it('returns an empty list when no webhooks are registered', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: [], error: null },
}),
)
const res = await listWebhooks(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data.webhooks).toEqual([])
})
})
// ──────────────────────────────────────────────────────────────────────
// GET /webhooks/:id (detail)
// ──────────────────────────────────────────────────────────────────────
describe('GET /api/v1/companies/:companyId/webhooks/:id', () => {
it('returns the webhook detail without secret', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: SAMPLE_WEBHOOK, error: null },
}),
)
const res = await getWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}`),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data.id).toBe(WEBHOOK_ID)
expect(body.data).not.toHaveProperty('secret')
})
it('returns 404 NOT_FOUND when the webhook does not exist for this company', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: null, error: null },
}),
)
const res = await getWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}`),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(404)
const body = await res.json()
expect(body.error.code).toBe('NOT_FOUND')
})
})
// ──────────────────────────────────────────────────────────────────────
// PATCH /webhooks/:id (update)
// ──────────────────────────────────────────────────────────────────────
describe('PATCH /api/v1/companies/:companyId/webhooks/:id', () => {
it('updates the webhook and clears disabled_at when active=true', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: {
data: { ...SAMPLE_WEBHOOK, disabled_at: null, disabled_reason: null },
error: null,
},
}),
)
const res = await updateWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ active: true }),
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data.disabled_at).toBeNull()
expect(body.data.disabled_reason).toBeNull()
})
it('re-runs the SSRF guard when webhook_url is changed', async () => {
mockUrlGuard.mockResolvedValueOnce({
ok: false,
reason: 'metadata_address',
detail: '169.254.169.254 is the cloud metadata endpoint',
})
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
}),
)
const res = await updateWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ webhook_url: 'https://metadata.example/hooks' }),
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
expect(body.error.details.reason).toBe('metadata_address')
})
it('returns 400 VALIDATION_ERROR for an empty body (no fields to update)', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
}),
)
const res = await updateWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({}),
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
})
})
// ──────────────────────────────────────────────────────────────────────
// DELETE /webhooks/:id
// ──────────────────────────────────────────────────────────────────────
describe('DELETE /api/v1/companies/:companyId/webhooks/:id', () => {
it('returns 204 NO_CONTENT after a successful delete', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: null, error: null },
}),
)
const res = await deleteWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}`, {
method: 'DELETE',
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(204)
})
})
// ──────────────────────────────────────────────────────────────────────
// POST /webhooks/:id/test (synthetic delivery)
// ──────────────────────────────────────────────────────────────────────
describe('POST /api/v1/companies/:companyId/webhooks/:id/test', () => {
it('enqueues a synthetic delivery and returns its id', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: {
data: { id: WEBHOOK_ID, api_version_pinned: '2026-05-12', active: true, disabled_at: null },
error: null,
},
webhook_deliveries: { data: { id: DELIVERY_ID }, error: null },
}),
)
const res = await testWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}/test`, {
method: 'POST',
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data.webhook_delivery_id).toBe(DELIVERY_ID)
expect(body.data.status).toBe('pending')
})
it('returns 404 NOT_FOUND when the webhook does not exist', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: null, error: null },
}),
)
const res = await testWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}/test`, {
method: 'POST',
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(404)
})
it('refuses to enqueue a test for a disabled webhook', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: {
data: {
id: WEBHOOK_ID,
api_version_pinned: '2026-05-12',
active: false,
disabled_at: '2026-05-15T11:00:00Z',
},
error: null,
},
}),
)
const res = await testWebhook(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}/test`, {
method: 'POST',
}),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
})
})
// ──────────────────────────────────────────────────────────────────────
// GET /webhooks/:id/deliveries (list deliveries)
// ──────────────────────────────────────────────────────────────────────
describe('GET /api/v1/companies/:companyId/webhooks/:id/deliveries', () => {
it('returns deliveries for the webhook with status + response details', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: { id: WEBHOOK_ID }, error: null },
webhook_deliveries: {
data: [
{
id: DELIVERY_ID,
webhook_id: WEBHOOK_ID,
event_type: 'invoice.paid',
status: 'delivered',
attempts: 1,
next_attempt_at: '2026-05-15T12:00:00Z',
response_status: 200,
response_body: 'ok',
error: null,
request_id: 'whfan_x',
created_at: '2026-05-15T12:00:00Z',
delivered_at: '2026-05-15T12:00:01Z',
},
],
error: null,
},
}),
)
const res = await listDeliveries(
makeRequest(
`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}/deliveries`,
),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data).toHaveLength(1)
expect(body.data[0].id).toBe(DELIVERY_ID)
expect(body.data[0].status).toBe('delivered')
expect(body.data[0].response_status).toBe(200)
})
it('returns 404 NOT_FOUND when the webhook does not exist for this company (clean signal vs empty list)', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
company_members: { data: { company_id: COMPANY_ID, role: 'owner' }, error: null },
webhooks: { data: null, error: null },
}),
)
const res = await listDeliveries(
makeRequest(
`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks/${WEBHOOK_ID}/deliveries`,
),
detailParams(COMPANY_ID, WEBHOOK_ID),
)
expect(res.status).toBe(404)
})
})
// ──────────────────────────────────────────────────────────────────────
// Cross-tenant URL guard (wrapper level)
// ──────────────────────────────────────────────────────────────────────
describe('webhook routes — cross-tenant URL guard', () => {
it('returns 404 NOT_FOUND when the caller is not a member of the company in the URL', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
// No membership row → wrapper short-circuits to NOT_FOUND.
company_members: { data: null, error: null },
}),
)
const res = await listWebhooks(
makeRequest(`https://x.test/api/v1/companies/${COMPANY_ID}/webhooks`),
companyParams(COMPANY_ID),
)
expect(res.status).toBe(404)
const body = await res.json()
expect(body.error.code).toBe('NOT_FOUND')
})
})
app/api/v1/companies/[companyId]/webhooks/route.ts
+5 -1
View File
@@ -298,10 +298,14 @@ export const POST = withApiV1<{ params: Promise<{ companyId: string }> }>(
const secret = `whsec_${generateWebhookSecret()}`
// No user_id field on the webhooks table — the column never existed
// in the automation_webhooks predecessor (20260415000000_schema_sync.sql)
// and webhooks_v2 (20260515170000) didn't add it. Actor attribution
// lives on created_by_api_key_id instead (which leads back to the
// owning user via api_keys.user_id).
const { data, error } = await ctx.supabase
.from('webhooks')
.insert({
user_id: ctx.userId,
company_id: ctx.companyId!,
name: body.name,
description: body.description ?? null,
app/api/v1/webhook-deliveries/[id]/retry/__tests__/route.test.ts
+327
View File
@@ -0,0 +1,327 @@
/**
* Integration tests for POST /api/v1/webhook-deliveries/:id/retry.
*
* The route lives outside the /companies/{companyId}/ tree (deliveries
* already carry their company_id; nesting would force callers to
* round-trip company resolution from the delivery id). Tenancy is still
* enforced — the route resolves the delivery's company_id, then verifies
* the caller is a member of that company via company_members.
*
* Closes the Phase 6 PR-1 (#496) integration-test debt for the retry
* route.
*/
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
beforeAll(() => {
if (process.env.NODE_ENV !== 'test') {
throw new Error(`retry route tests require NODE_ENV=test`)
}
process.env.NEXT_PUBLIC_SUPABASE_URL ||= 'http://localhost:54321'
process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY ||= 'test-anon-key'
})
vi.mock('@/lib/auth/api-keys', async () => {
const actual = await vi.importActual<typeof import('@/lib/auth/api-keys')>('@/lib/auth/api-keys')
return {
...actual,
validateApiKey: vi.fn(),
createServiceClientNoCookies: vi.fn(),
}
})
vi.mock('@supabase/supabase-js', async () => {
const actual = await vi.importActual<typeof import('@supabase/supabase-js')>('@supabase/supabase-js')
return { ...actual, createClient: vi.fn().mockReturnValue({}) }
})
vi.mock('@/lib/webhooks/url-guard', async () => {
const actual = await vi.importActual<typeof import('@/lib/webhooks/url-guard')>(
'@/lib/webhooks/url-guard',
)
return {
...actual,
validateWebhookUrl: vi.fn(),
}
})
import { validateApiKey, createServiceClientNoCookies } from '@/lib/auth/api-keys'
import { validateWebhookUrl } from '@/lib/webhooks/url-guard'
import { POST as retryDelivery } from '../route'
const mockValidate = validateApiKey as ReturnType<typeof vi.fn>
const mockServiceClient = createServiceClientNoCookies as ReturnType<typeof vi.fn>
const mockUrlGuard = validateWebhookUrl as ReturnType<typeof vi.fn>
interface TableResp {
data?: unknown
error?: unknown
}
function makeFlexibleSupabase(byTable: Record<string, TableResp | TableResp[]>) {
const queues = new Map<string, TableResp[]>()
for (const [t, val] of Object.entries(byTable)) {
queues.set(t, Array.isArray(val) ? [...val] : [val])
}
const buildChain = (table: string): unknown => {
const handler: ProxyHandler<object> = {
get(_target, prop) {
if (prop === 'then') {
return (resolve: (v: unknown) => void) => {
const q = queues.get(table)
const next = q && q.length > 1 ? q.shift()! : (q?.[0] ?? { data: null, error: null })
resolve(next)
}
}
return (..._args: unknown[]) => buildChain(table)
},
}
return new Proxy({}, handler)
}
return { from: vi.fn((table: string) => buildChain(table)) }
}
const COMPANY_ID = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'
const WEBHOOK_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'
const DELIVERY_ID = 'cccccccc-cccc-4ccc-8ccc-cccccccccccc'
const NEW_DELIVERY_ID = 'dddddddd-dddd-4ddd-8ddd-dddddddddddd'
const USER_ID = 'user-1'
function makeRequest(url: string, init?: RequestInit): Request {
return new Request(url, {
...init,
headers: {
Authorization: 'Bearer test-fixture-not-a-real-key',
...(init?.headers ?? {}),
},
})
}
function idParams(id: string) {
return { params: Promise.resolve({ id }) }
}
const DEAD_DELIVERY = {
id: DELIVERY_ID,
webhook_id: WEBHOOK_ID,
company_id: COMPANY_ID,
event_type: 'invoice.paid',
payload: { invoice_id: 'inv_x' },
previous_attributes: null,
api_version: '2026-05-12',
status: 'dead' as const,
}
const ACTIVE_WEBHOOK = {
id: WEBHOOK_ID,
webhook_url: 'https://example.com/hooks',
active: true,
disabled_at: null,
}
beforeEach(() => {
vi.clearAllMocks()
mockValidate.mockResolvedValue({
userId: USER_ID,
companyId: COMPANY_ID,
apiKeyId: 'ak_1',
apiKeyName: 'CI key',
scopes: ['webhooks:manage'],
mode: 'live',
})
mockUrlGuard.mockResolvedValue({
ok: true,
hostname: 'example.com',
resolvedAddresses: ['203.0.113.42'],
})
})
describe('POST /api/v1/webhook-deliveries/:id/retry', () => {
it('re-enqueues a dead delivery as a fresh pending row', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: [
{ data: DEAD_DELIVERY, error: null }, // lookup
{ data: { id: NEW_DELIVERY_ID }, error: null }, // insert
],
company_members: { data: { company_id: COMPANY_ID }, error: null },
webhooks: { data: ACTIVE_WEBHOOK, error: null },
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(200)
const body = await res.json()
expect(body.data.webhook_delivery_id).toBe(NEW_DELIVERY_ID)
expect(body.data.status).toBe('pending')
})
it('requires payroll:read for salary_run.* / agi.* retries (elevated-scope gate)', async () => {
// Caller has webhooks:manage but NOT payroll:read. Original create
// would have rejected the subscription; retry must reject the
// re-emission identically so a stripped-down key can't replay payroll
// payloads to its receiver.
mockValidate.mockResolvedValueOnce({
userId: USER_ID,
companyId: COMPANY_ID,
apiKeyId: 'ak_1',
apiKeyName: 'CI key',
scopes: ['webhooks:manage'],
mode: 'live',
})
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: {
data: { ...DEAD_DELIVERY, event_type: 'salary_run.booked' },
error: null,
},
company_members: { data: { company_id: COMPANY_ID }, error: null },
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(403)
const body = await res.json()
expect(body.error.code).toBe('INSUFFICIENT_SCOPE')
expect(body.error.details.required_scope).toBe('payroll:read')
})
it('refuses to retry a live delivery (pending/in_flight/failed)', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: { data: { ...DEAD_DELIVERY, status: 'failed' }, error: null },
company_members: { data: { company_id: COMPANY_ID }, error: null },
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
expect(body.error.details.message).toMatch(/dead or delivered/i)
})
it('returns 404 NOT_FOUND when the caller is not a member of the delivery company', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: { data: DEAD_DELIVERY, error: null },
// Non-member → 404, not 403, so we don't leak delivery existence.
company_members: { data: null, error: null },
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(404)
})
it('refuses to retry against a disabled webhook', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: { data: DEAD_DELIVERY, error: null },
company_members: { data: { company_id: COMPANY_ID }, error: null },
webhooks: {
data: { ...ACTIVE_WEBHOOK, active: false, disabled_at: '2026-05-15T11:00:00Z' },
error: null,
},
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
})
it('refuses to retry when the webhook URL fails the SSRF re-check', async () => {
mockUrlGuard.mockResolvedValueOnce({
ok: false,
reason: 'private_address',
detail: '10.0.0.1 is private',
})
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: { data: DEAD_DELIVERY, error: null },
company_members: { data: { company_id: COMPANY_ID }, error: null },
webhooks: { data: ACTIVE_WEBHOOK, error: null },
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(400)
const body = await res.json()
expect(body.error.code).toBe('VALIDATION_ERROR')
expect(body.error.details.reason).toBe('private_address')
})
it('returns 404 NOT_FOUND when the delivery does not exist', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: { data: null, error: null },
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(404)
})
it('returns 404 NOT_FOUND when the original webhook has been deleted', async () => {
mockServiceClient.mockReturnValue(
makeFlexibleSupabase({
webhook_deliveries: { data: DEAD_DELIVERY, error: null },
company_members: { data: { company_id: COMPANY_ID }, error: null },
webhooks: { data: null, error: null }, // webhook deleted between dead and retry
}),
)
const res = await retryDelivery(
makeRequest(`https://x.test/api/v1/webhook-deliveries/${DELIVERY_ID}/retry`, {
method: 'POST',
}),
idParams(DELIVERY_ID),
)
expect(res.status).toBe(404)
})
})
app/api/v1/webhook-deliveries/[id]/retry/route.ts
+18
View File
@@ -25,6 +25,7 @@ import { withApiV1 } from '@/lib/api/v1/with-api-v1'
import { v1ErrorResponse, v1ErrorResponseFromCode } from '@/lib/api/v1/errors'
import { minimisePayload } from '@/lib/webhooks/handler'
import { validateWebhookUrl } from '@/lib/webhooks/url-guard'
import { hasScope } from '@/lib/auth/api-keys'
registerEndpoint({
operation: 'webhook_deliveries.retry',
@@ -119,6 +120,23 @@ export const POST = withApiV1<{ params: Promise<{ id: string }> }>(
})
}
// Mirror the create-route elevated-scope gate. A key with only
// webhooks:manage must NOT be able to re-emit a salary_run.* / agi.*
// payload — those carry personnummer, lönesummor, skatteavdrag, and
// the original create call required webhooks:manage AND payroll:read.
// Retry checks the SAME pair against the CALLING key's scopes (which
// may differ from the key that created the webhook in the first place).
const PAYROLL_SENSITIVE = /^(salary_run\.|agi\.)/
if (PAYROLL_SENSITIVE.test(o.event_type) && !hasScope(ctx.scopes, 'payroll:read')) {
return v1ErrorResponseFromCode('INSUFFICIENT_SCOPE', ctx.log, {
requestId: ctx.requestId,
details: {
required_scope: 'payroll:read',
reason: `Retrying ${o.event_type} requires payroll:read in addition to webhooks:manage.`,
},
})
}
// Re-verify that the parent webhook still exists, still belongs to the
// delivery's company, and is still active immediately before INSERT.
// Closes the TOCTOU window between the membership check above and the
lib/api/v1/__tests__/operations-immutability.pg.test.ts
+171
View File
@@ -0,0 +1,171 @@
import { randomUUID } from 'node:crypto'
import { describe, expect, it } from 'vitest'
import { getPool } from '@/tests/pg/setup'
import { seedCompany } from '@/tests/pg/fixtures'
type OperationStatus = 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled'
async function insertOperation(params: {
companyId: string
userId: string
status: OperationStatus
result?: Record<string, unknown> | null
error?: Record<string, unknown> | null
}): Promise<string> {
const id = randomUUID()
const isTerminal =
params.status === 'succeeded' ||
params.status === 'failed' ||
params.status === 'cancelled'
await getPool().query(
`INSERT INTO public.operations
(id, company_id, user_id, operation_type, status, started_at, completed_at, result, error)
VALUES ($1, $2, $3, 'test.op', $4, now(), $5, $6, $7)`,
[
id,
params.companyId,
params.userId,
params.status,
isTerminal ? new Date() : null,
params.result ? JSON.stringify(params.result) : null,
params.error ? JSON.stringify(params.error) : null,
],
)
return id
}
describe('operations-immutability.pg — terminal-status rows are immutable', () => {
// Sanity: the regular running → succeeded transition (the path used by
// completeOperation in lib/api/v1/operations.ts) is NOT blocked. The
// trigger reads OLD.status; the legitimate UPDATE has OLD.status='running'
// which is non-terminal, so the RAISE is skipped.
it('allows UPDATE that transitions running → succeeded', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({ companyId, userId, status: 'running' })
await expect(
getPool().query(
`UPDATE public.operations
SET status = 'succeeded',
completed_at = now(),
result = '{"ok": true}'::jsonb
WHERE id = $1`,
[opId],
),
).resolves.toMatchObject({ rowCount: 1 })
})
it('allows UPDATE that transitions queued → running', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({ companyId, userId, status: 'queued' })
await expect(
getPool().query(
`UPDATE public.operations
SET status = 'running', started_at = now()
WHERE id = $1`,
[opId],
),
).resolves.toMatchObject({ rowCount: 1 })
})
it('rejects UPDATE of result on a succeeded row', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({
companyId,
userId,
status: 'succeeded',
result: { ok: true },
})
await expect(
getPool().query(
`UPDATE public.operations SET result = '{"tampered": true}'::jsonb WHERE id = $1`,
[opId],
),
).rejects.toThrow(/terminal status \(succeeded\) is immutable/i)
})
it('rejects UPDATE of status on a failed row (no failed → running reopen)', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({
companyId,
userId,
status: 'failed',
error: { code: 'X' },
})
await expect(
getPool().query(
`UPDATE public.operations SET status = 'running' WHERE id = $1`,
[opId],
),
).rejects.toThrow(/terminal status \(failed\) is immutable/i)
})
it('rejects UPDATE of error on a cancelled row', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({ companyId, userId, status: 'cancelled' })
await expect(
getPool().query(
`UPDATE public.operations SET error = '{"code": "REWRITTEN"}'::jsonb WHERE id = $1`,
[opId],
),
).rejects.toThrow(/terminal status \(cancelled\) is immutable/i)
})
it('rejects DELETE on a succeeded row', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({
companyId,
userId,
status: 'succeeded',
result: { ok: true },
})
await expect(
getPool().query(`DELETE FROM public.operations WHERE id = $1`, [opId]),
).rejects.toThrow(/terminal status \(succeeded\) cannot be deleted/i)
})
it('rejects DELETE on a failed row', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({
companyId,
userId,
status: 'failed',
error: { code: 'X' },
})
await expect(
getPool().query(`DELETE FROM public.operations WHERE id = $1`, [opId]),
).rejects.toThrow(/terminal status \(failed\) cannot be deleted/i)
})
// Non-terminal rows remain deletable so operators can clear stuck/queued
// entries (e.g. a worker that crashed before claiming the row, manual
// cleanup of dev/test environments). The retention obligation kicks in
// only once the row has recorded a terminal outcome.
it('allows DELETE on a queued row', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({ companyId, userId, status: 'queued' })
const result = await getPool().query(
`DELETE FROM public.operations WHERE id = $1`,
[opId],
)
expect(result.rowCount).toBe(1)
})
it('allows DELETE on a running row', async () => {
const { userId, companyId } = await seedCompany()
const opId = await insertOperation({ companyId, userId, status: 'running' })
const result = await getPool().query(
`DELETE FROM public.operations WHERE id = $1`,
[opId],
)
expect(result.rowCount).toBe(1)
})
})
lib/webhooks/__tests__/claim-due-webhook-deliveries.pg.test.ts
+284
View File
@@ -0,0 +1,284 @@
import { randomUUID } from 'node:crypto'
import { describe, expect, it } from 'vitest'
import { getClient, getPool } from '@/tests/pg/setup'
import { seedCompany } from '@/tests/pg/fixtures'
// ──────────────────────────────────────────────────────────────────────
// Fixtures: parent webhook + child delivery
// ──────────────────────────────────────────────────────────────────────
async function insertWebhook(params: {
// userId kept in the signature for parity with seedCompany's return — the
// webhooks table itself has no user_id column (see route comment in
// app/api/v1/companies/[companyId]/webhooks/route.ts).
userId: string
companyId: string
eventType?: string
active?: boolean
}): Promise<string> {
void params.userId
const id = randomUUID()
await getPool().query(
`INSERT INTO public.webhooks
(id, company_id, name, event_type, webhook_url, secret, active)
VALUES ($1, $2, 'pg-test', $3, 'https://example.com/hook', $4, $5)`,
[
id,
params.companyId,
params.eventType ?? 'invoice.paid',
`whsec_${randomUUID().replace(/-/g, '')}`,
params.active ?? true,
],
)
return id
}
async function insertDelivery(params: {
webhookId: string | null
companyId: string
status?: 'pending' | 'in_flight' | 'delivered' | 'failed' | 'dead'
nextAttemptAt?: string
eventType?: string
attempts?: number
}): Promise<string> {
const id = randomUUID()
await getPool().query(
`INSERT INTO public.webhook_deliveries
(id, webhook_id, company_id, event_type, payload, api_version,
status, next_attempt_at, attempts)
VALUES ($1, $2, $3, $4, '{"hello":"world"}'::jsonb, '2026-05-12',
$5, $6, $7)`,
[
id,
params.webhookId,
params.companyId,
params.eventType ?? 'invoice.paid',
params.status ?? 'pending',
params.nextAttemptAt ?? new Date().toISOString(),
params.attempts ?? 0,
],
)
return id
}
async function getDeliveryStatus(id: string): Promise<string | null> {
const r = await getPool().query<{ status: string }>(
`SELECT status FROM public.webhook_deliveries WHERE id = $1`,
[id],
)
return r.rows[0]?.status ?? null
}
// Direct-insert rows for one test isolation. Each it() seeds its own
// company + webhook so the dispatcher sees a clean slate; we just need to
// make sure the function only returns rows we created, which we do by
// asserting on ids.
describe('claim_due_webhook_deliveries.pg — atomic SKIP LOCKED claim', () => {
it('claims a pending row that is due', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'pending' })
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(rows.map((r) => r.id)).toContain(deliveryId)
expect(await getDeliveryStatus(deliveryId)).toBe('in_flight')
})
it('claims a failed row that has reached its retry deadline', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
// next_attempt_at in the past — retry due.
const deliveryId = await insertDelivery({
webhookId,
companyId,
status: 'failed',
nextAttemptAt: new Date(Date.now() - 60_000).toISOString(),
attempts: 2,
})
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(rows.map((r) => r.id)).toContain(deliveryId)
expect(await getDeliveryStatus(deliveryId)).toBe('in_flight')
})
it('skips a row whose next_attempt_at is still in the future', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({
webhookId,
companyId,
status: 'failed',
nextAttemptAt: new Date(Date.now() + 5 * 60_000).toISOString(),
})
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(rows.map((r) => r.id)).not.toContain(deliveryId)
// Status stays pre-claim.
expect(await getDeliveryStatus(deliveryId)).toBe('failed')
})
it('skips dangling rows (webhook_id IS NULL)', async () => {
const { companyId } = await seedCompany()
// webhook deleted between enqueue and dispatch — FK ON DELETE SET NULL.
const deliveryId = await insertDelivery({
webhookId: null,
companyId,
status: 'pending',
})
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(rows.map((r) => r.id)).not.toContain(deliveryId)
})
it('skips terminal-status rows', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveredId = await insertDelivery({
webhookId,
companyId,
status: 'delivered',
})
const deadId = await insertDelivery({
webhookId,
companyId,
status: 'dead',
})
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
const ids = rows.map((r) => r.id)
expect(ids).not.toContain(deliveredId)
expect(ids).not.toContain(deadId)
})
// The status filter `IN ('pending', 'failed')` is what prevents
// double-delivery once a tick has already claimed a row to in_flight.
// recoverStuckInFlight is the ONLY legitimate path back from in_flight
// (sweeps the row to 'failed' after the stuck-threshold), so the claim
// function must NEVER re-pick a row already marked in_flight.
it('skips rows already in in_flight status', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const inFlightId = await insertDelivery({
webhookId,
companyId,
status: 'in_flight',
})
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(rows.map((r) => r.id)).not.toContain(inFlightId)
// Status must NOT have been re-flipped.
expect(await getDeliveryStatus(inFlightId)).toBe('in_flight')
})
it('respects the batch size', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const ids = await Promise.all(
Array.from({ length: 5 }, () =>
insertDelivery({ webhookId, companyId, status: 'pending' }),
),
)
const { rows } = await getPool().query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[2],
)
expect(rows.length).toBe(2)
// The 3 unclaimed rows stay pending.
const unclaimed = ids.filter((id) => !rows.some((r) => r.id === id))
for (const id of unclaimed) {
expect(await getDeliveryStatus(id)).toBe('pending')
}
})
it('rejects out-of-range batch sizes', async () => {
await expect(
getPool().query(`SELECT * FROM public.claim_due_webhook_deliveries($1, now())`, [0]),
).rejects.toThrow(/p_batch_size must be in/i)
await expect(
getPool().query(`SELECT * FROM public.claim_due_webhook_deliveries($1, now())`, [-1]),
).rejects.toThrow(/p_batch_size must be in/i)
await expect(
getPool().query(`SELECT * FROM public.claim_due_webhook_deliveries($1, now())`, [10000]),
).rejects.toThrow(/p_batch_size must be in/i)
})
// SKIP LOCKED is the entire point of this migration. Two transactions
// calling the function at the same moment must not both see the same
// row — the row locked by the first caller is invisible to the second,
// closing the duplicate-delivery window the old SELECT-then-UPDATE-
// intersect pattern documented as load-bearing.
it('SKIP LOCKED: a concurrent caller does not see rows locked by an in-flight transaction', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({
webhookId,
companyId,
status: 'pending',
})
const a = await getClient()
const b = await getClient()
try {
await a.query('BEGIN')
await b.query('BEGIN')
// A claims first. The row is now status='in_flight' AND held under
// a row lock by transaction A (UPDATE sets ROW EXCLUSIVE).
const aClaim = await a.query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(aClaim.rows.map((r) => r.id)).toContain(deliveryId)
// B's call SKIPs the locked row entirely. Without SKIP LOCKED this
// call would BLOCK on the row lock; the test would hang and only
// fail via testTimeout. SKIP LOCKED makes it return promptly with
// the row simply absent from results.
const bClaim = await b.query<{ id: string }>(
`SELECT id FROM public.claim_due_webhook_deliveries($1, now())`,
[10],
)
expect(bClaim.rows.map((r) => r.id)).not.toContain(deliveryId)
// Commit A and roll back B (no-op since B claimed nothing).
await a.query('COMMIT')
await b.query('ROLLBACK')
} finally {
a.release()
b.release()
}
// Final state: row is in_flight, claimed exactly once.
expect(await getDeliveryStatus(deliveryId)).toBe('in_flight')
})
})
lib/webhooks/__tests__/pinned-fetch.test.ts
+271
View File
@@ -0,0 +1,271 @@
import { describe, expect, it, vi } from 'vitest'
import { EventEmitter } from 'node:events'
import type { ClientRequest, IncomingMessage } from 'node:http'
import type { RequestOptions } from 'node:https'
import { pinnedHttpsFetch } from '@/lib/webhooks/pinned-fetch'
// Stand up a minimal stub for node:https.request that captures the args
// we want to assert on (pinned IP, SNI, Host header) and lets us synthesise
// a controlled response back to the caller.
function makeStubRequest(args: {
status: number
headers?: Record<string, string>
body?: string
emitError?: Error
emitTimeout?: boolean
}) {
const captured: {
options: RequestOptions | null
bodyWritten: string
} = { options: null, bodyWritten: '' }
const fakeRequest = (
options: RequestOptions,
callback: (res: IncomingMessage) => void,
): ClientRequest => {
captured.options = options
const req = new EventEmitter() as ClientRequest & EventEmitter
// ClientRequest API surface we touch in pinned-fetch:
req.write = ((chunk: string) => {
captured.bodyWritten += chunk
return true
}) as ClientRequest['write']
req.end = (() => {
// Dispatch the response (or error) asynchronously to mimic real
// network timing — pinned-fetch attaches handlers BEFORE end().
queueMicrotask(() => {
if (args.emitError) {
req.emit('error', args.emitError)
return
}
if (args.emitTimeout) {
req.emit('timeout')
return
}
const res = new EventEmitter() as IncomingMessage & EventEmitter
;(res as unknown as { statusCode: number }).statusCode = args.status
;(res as unknown as { headers: Record<string, string> }).headers =
args.headers ?? { 'content-type': 'application/json' }
// IncomingMessage stubs need stream-shaped methods that pinned-fetch
// calls (resume on redirect-drain, destroy on size truncation).
;(res as unknown as { resume: () => unknown }).resume = () => {
/* no-op — body is already buffered in args.body */
}
;(res as unknown as { destroy: () => unknown }).destroy = () => {
// Truncation path — emit `close` so finalize() runs.
queueMicrotask(() => res.emit('close'))
}
callback(res)
// Emit body bytes then `end`.
queueMicrotask(() => {
if (args.body) res.emit('data', Buffer.from(args.body, 'utf8'))
res.emit('end')
})
})
return req
}) as ClientRequest['end']
req.destroy = (() => {
// no-op: tests don't read the socket after destroy.
return req
}) as ClientRequest['destroy']
req.setTimeout = (() => req) as ClientRequest['setTimeout']
return req
}
return { captured, fakeRequest }
}
function makeStubValidator(addresses: string[]) {
return vi.fn(async () => ({
ok: true as const,
hostname: 'example.com',
resolvedAddresses: addresses,
}))
}
describe('pinnedHttpsFetch', () => {
it('pins the socket to the validated IP while keeping SNI + Host on the hostname', async () => {
const { captured, fakeRequest } = makeStubRequest({
status: 200,
body: 'ok',
})
const result = await pinnedHttpsFetch(
'https://example.com/hooks',
{
method: 'POST',
headers: { 'X-Gnubok-Event': 'invoice.paid' },
body: '{"hello":"world"}',
timeoutMs: 1000,
maxResponseBytes: 1024,
},
{
validateUrl: makeStubValidator(['203.0.113.42']),
httpsRequest: fakeRequest,
},
)
expect(result.kind).toBe('ok')
if (result.kind !== 'ok') throw new Error('unreachable')
expect(result.status).toBe(200)
expect(result.body).toBe('ok')
expect(result.pinnedAddress).toBe('203.0.113.42')
// Socket goes to the IP — DNS does not re-resolve.
expect(captured.options?.host).toBe('203.0.113.42')
// SNI carries the hostname so the receiver's TLS cert validates.
expect(captured.options?.servername).toBe('example.com')
// HTTP Host header carries the hostname for vhost routing.
const headers = captured.options?.headers as Record<string, string>
expect(headers.host).toBe('example.com')
// Custom dispatcher header survives.
expect(headers['X-Gnubok-Event']).toBe('invoice.paid')
expect(captured.bodyWritten).toBe('{"hello":"world"}')
})
it('includes the port in the Host header when non-default', async () => {
const { captured, fakeRequest } = makeStubRequest({ status: 204 })
await pinnedHttpsFetch(
'https://example.com:8443/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 1024 },
{
validateUrl: makeStubValidator(['203.0.113.42']),
httpsRequest: fakeRequest,
},
)
expect(captured.options?.port).toBe(8443)
const headers = captured.options?.headers as Record<string, string>
expect(headers.host).toBe('example.com:8443')
})
it('returns unsafe_url when validation rejects the hostname', async () => {
const { fakeRequest } = makeStubRequest({ status: 200 })
const spyRequest = vi.fn(fakeRequest)
const validator = vi.fn(async () => ({
ok: false as const,
reason: 'private_address' as const,
detail: '10.0.0.1 is private',
}))
const result = await pinnedHttpsFetch(
'https://internal.example/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 1024 },
{ validateUrl: validator, httpsRequest: spyRequest },
)
expect(result.kind).toBe('unsafe_url')
if (result.kind === 'unsafe_url') {
expect(result.reason).toBe('private_address')
expect(result.pinnedAddress).toBeNull()
}
// Critically — we never opened a socket.
expect(spyRequest).not.toHaveBeenCalled()
})
it('treats 3xx responses as redirect_blocked', async () => {
const { fakeRequest } = makeStubRequest({
status: 302,
headers: { location: 'https://elsewhere.example/' },
})
const result = await pinnedHttpsFetch(
'https://example.com/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 1024 },
{
validateUrl: makeStubValidator(['203.0.113.42']),
httpsRequest: fakeRequest,
},
)
expect(result.kind).toBe('redirect_blocked')
if (result.kind === 'redirect_blocked') {
expect(result.status).toBe(302)
expect(result.pinnedAddress).toBe('203.0.113.42')
}
})
it('maps transport errors to transport_error', async () => {
const { fakeRequest } = makeStubRequest({
status: 0,
emitError: new Error('ECONNREFUSED 203.0.113.42:443'),
})
const result = await pinnedHttpsFetch(
'https://example.com/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 1024 },
{
validateUrl: makeStubValidator(['203.0.113.42']),
httpsRequest: fakeRequest,
},
)
expect(result.kind).toBe('transport_error')
if (result.kind === 'transport_error') {
expect(result.detail).toContain('ECONNREFUSED')
expect(result.pinnedAddress).toBe('203.0.113.42')
}
})
it('maps timeout events to timeout', async () => {
const { fakeRequest } = makeStubRequest({
status: 0,
emitTimeout: true,
})
const result = await pinnedHttpsFetch(
'https://example.com/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 1024 },
{
validateUrl: makeStubValidator(['203.0.113.42']),
httpsRequest: fakeRequest,
},
)
expect(result.kind).toBe('timeout')
})
it('truncates response body at maxResponseBytes', async () => {
const big = 'x'.repeat(10_000)
const { fakeRequest } = makeStubRequest({
status: 200,
headers: { 'content-type': 'text/plain' },
body: big,
})
const result = await pinnedHttpsFetch(
'https://example.com/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 100 },
{
validateUrl: makeStubValidator(['203.0.113.42']),
httpsRequest: fakeRequest,
},
)
expect(result.kind).toBe('ok')
if (result.kind === 'ok') {
expect(result.body.length).toBe(100)
expect(result.bodyTruncated).toBe(true)
}
})
it('picks the first resolved IP deterministically', async () => {
const { captured, fakeRequest } = makeStubRequest({ status: 200 })
await pinnedHttpsFetch(
'https://example.com/hooks',
{ method: 'POST', headers: {}, body: '', timeoutMs: 1000, maxResponseBytes: 1024 },
{
validateUrl: makeStubValidator(['203.0.113.42', '198.51.100.55']),
httpsRequest: fakeRequest,
},
)
// First entry, not the second, not random.
expect(captured.options?.host).toBe('203.0.113.42')
})
})
lib/webhooks/__tests__/webhook-triggers.pg.test.ts
+261
View File
@@ -0,0 +1,261 @@
import { randomUUID } from 'node:crypto'
import { describe, expect, it } from 'vitest'
import { getPool } from '@/tests/pg/setup'
import { seedCompany } from '@/tests/pg/fixtures'
// Verifies the three webhook-substrate DB guards shipped in Phase 6 PR-1:
// - enforce_webhook_delivery_immutability (BEFORE UPDATE)
// - block_webhook_delivery_terminal_delete (BEFORE DELETE)
// - assert_webhook_delivery_company_match (BEFORE INSERT)
//
// CLAUDE.md ("Migration Rules" + Testing section) mandates a *.pg.test.ts
// for any PR that touches a trigger / RPC / RLS / DEFERRABLE constraint.
// PR-1 (#496) shipped the triggers without the accompanying pg test; this
// closes that test debt.
async function insertWebhook(params: {
// userId kept in the signature for parity with seedCompany's return — the
// webhooks table itself has no user_id column (see route comment in
// app/api/v1/companies/[companyId]/webhooks/route.ts).
userId: string
companyId: string
eventType?: string
}): Promise<string> {
void params.userId
const id = randomUUID()
await getPool().query(
`INSERT INTO public.webhooks
(id, company_id, name, event_type, webhook_url, secret, active)
VALUES ($1, $2, 'pg-test', $3, 'https://example.com/hook', $4, true)`,
[
id,
params.companyId,
params.eventType ?? 'invoice.paid',
`whsec_${randomUUID().replace(/-/g, '')}`,
],
)
return id
}
async function insertDelivery(params: {
webhookId: string | null
companyId: string
status?: 'pending' | 'in_flight' | 'delivered' | 'failed' | 'dead'
}): Promise<string> {
const id = randomUUID()
await getPool().query(
`INSERT INTO public.webhook_deliveries
(id, webhook_id, company_id, event_type, payload, api_version,
status, next_attempt_at)
VALUES ($1, $2, $3, 'invoice.paid', '{"hello":"world"}'::jsonb,
'2026-05-12', $4, now())`,
[id, params.webhookId, params.companyId, params.status ?? 'pending'],
)
return id
}
describe('webhook_deliveries triggers — immutability + DELETE block', () => {
// The lifecycle that dispatcher.ts depends on must remain mutable:
// pending → in_flight (claim), in_flight → failed (retry-pending),
// failed → in_flight (re-claim). Only `delivered` and `dead` are
// terminal and locked.
it('allows pending → in_flight (claim) transition', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'pending' })
const result = await getPool().query(
`UPDATE public.webhook_deliveries SET status = 'in_flight' WHERE id = $1`,
[deliveryId],
)
expect(result.rowCount).toBe(1)
})
it('allows in_flight → failed (retry-pending) transition', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'in_flight' })
const result = await getPool().query(
`UPDATE public.webhook_deliveries SET status = 'failed', attempts = 1 WHERE id = $1`,
[deliveryId],
)
expect(result.rowCount).toBe(1)
})
it('allows failed → in_flight (re-claim) transition', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'failed' })
const result = await getPool().query(
`UPDATE public.webhook_deliveries SET status = 'in_flight' WHERE id = $1`,
[deliveryId],
)
expect(result.rowCount).toBe(1)
})
it('allows in_flight → delivered (success terminal) transition', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'in_flight' })
const result = await getPool().query(
`UPDATE public.webhook_deliveries
SET status = 'delivered', delivered_at = now(), response_status = 200
WHERE id = $1`,
[deliveryId],
)
expect(result.rowCount).toBe(1)
})
it('rejects UPDATE on a delivered row (status flip-back blocked)', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'delivered' })
await expect(
getPool().query(
`UPDATE public.webhook_deliveries SET status = 'pending' WHERE id = $1`,
[deliveryId],
),
).rejects.toThrow(/terminal status \(delivered\) is immutable/i)
})
it('rejects UPDATE on a dead row (response rewrite blocked)', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'dead' })
await expect(
getPool().query(
`UPDATE public.webhook_deliveries
SET response_body = 'tampered' WHERE id = $1`,
[deliveryId],
),
).rejects.toThrow(/terminal status \(dead\) is immutable/i)
})
it('rejects DELETE on a delivered row', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'delivered' })
await expect(
getPool().query(
`DELETE FROM public.webhook_deliveries WHERE id = $1`,
[deliveryId],
),
).rejects.toThrow(/terminal status \(delivered\) cannot be deleted/i)
})
it('rejects DELETE on a dead row', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'dead' })
await expect(
getPool().query(
`DELETE FROM public.webhook_deliveries WHERE id = $1`,
[deliveryId],
),
).rejects.toThrow(/terminal status \(dead\) cannot be deleted/i)
})
// Non-terminal rows are still deletable — the queue-cleanup path
// (operator clears a stuck pending row, dev environment wipes,
// companies CASCADE delete) keeps working.
it('allows DELETE on a pending row', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const deliveryId = await insertDelivery({ webhookId, companyId, status: 'pending' })
const result = await getPool().query(
`DELETE FROM public.webhook_deliveries WHERE id = $1`,
[deliveryId],
)
expect(result.rowCount).toBe(1)
})
})
describe('webhook_deliveries triggers — cross-tenant INSERT guard', () => {
it('rejects INSERT when delivery.company_id != webhooks.company_id', async () => {
// Tenant A owns the webhook; tenant B owns the company on the
// delivery row. A compromised service-role caller (or future bug)
// attempting to enqueue a delivery against another tenant's webhook
// is refused at write time — closes cases (a) (cross-tenant
// visibility) and (c) (existence leak) flagged in migration
// 20260515190000's comment.
const a = await seedCompany()
const b = await seedCompany()
const webhookId = await insertWebhook({ userId: a.userId, companyId: a.companyId })
await expect(
getPool().query(
`INSERT INTO public.webhook_deliveries
(id, webhook_id, company_id, event_type, payload, api_version,
status, next_attempt_at)
VALUES (gen_random_uuid(), $1, $2, 'invoice.paid',
'{"hello":"world"}'::jsonb, '2026-05-12', 'pending', now())`,
[webhookId, b.companyId],
),
).rejects.toThrow(/company_id .+ does not match parent webhooks\.company_id/i)
})
it('accepts INSERT when delivery.company_id == webhooks.company_id', async () => {
const { userId, companyId } = await seedCompany()
const webhookId = await insertWebhook({ userId, companyId })
const result = await getPool().query(
`INSERT INTO public.webhook_deliveries
(id, webhook_id, company_id, event_type, payload, api_version,
status, next_attempt_at)
VALUES (gen_random_uuid(), $1, $2, 'invoice.paid',
'{"hello":"world"}'::jsonb, '2026-05-12', 'pending', now())
RETURNING id`,
[webhookId, companyId],
)
expect(result.rowCount).toBe(1)
})
it('allows INSERT with webhook_id IS NULL (dangling row after webhook delete)', async () => {
// ON DELETE SET NULL on the FK leaves these rows after a webhook is
// deleted. New inserts with webhook_id IS NULL aren't a normal write
// path (the handler never inserts a null webhook_id) but the trigger
// explicitly bypasses the check rather than blocking — leaving room
// for an admin-side audit-replay tool that recreates an archived
// delivery for forensic export.
const { companyId } = await seedCompany()
const result = await getPool().query(
`INSERT INTO public.webhook_deliveries
(id, webhook_id, company_id, event_type, payload, api_version,
status, next_attempt_at)
VALUES (gen_random_uuid(), NULL, $1, 'invoice.paid',
'{"hello":"world"}'::jsonb, '2026-05-12', 'pending', now())
RETURNING id`,
[companyId],
)
expect(result.rowCount).toBe(1)
})
it('rejects INSERT pointing at a non-existent webhook_id', async () => {
// Bypass-check path: the trigger early-returns when the parent
// lookup yields NULL so the FK constraint surfaces the bad
// reference rather than the more-confusing company_match error.
// This test pins the FK-error pathway.
const { companyId } = await seedCompany()
const ghostWebhookId = randomUUID()
await expect(
getPool().query(
`INSERT INTO public.webhook_deliveries
(id, webhook_id, company_id, event_type, payload, api_version,
status, next_attempt_at)
VALUES (gen_random_uuid(), $1, $2, 'invoice.paid',
'{"hello":"world"}'::jsonb, '2026-05-12', 'pending', now())`,
[ghostWebhookId, companyId],
),
).rejects.toThrow(/webhook_deliveries_webhook_id_fkey|foreign key/i)
})
})
lib/webhooks/dispatcher.ts
+125 -203
View File
@@ -23,7 +23,7 @@
import type { SupabaseClient } from '@supabase/supabase-js'
import { signPayload } from './signing'
import { validateWebhookUrl } from './url-guard'
import { pinnedHttpsFetch, type PinnedFetchResult } from './pinned-fetch'
import { createLogger } from '@/lib/logger'
const log = createLogger('webhooks/dispatcher')
@@ -80,12 +80,12 @@ export async function dispatchDueDeliveries(args: {
batchSize?: number
/** Override for tests. */
now?: Date
/** Override for tests; injected fetch implementation. */
fetchImpl?: typeof fetch
/** Override for tests; injected pinned-fetch implementation. */
pinnedFetchImpl?: typeof pinnedHttpsFetch
}): Promise<DispatchSummary> {
const batchSize = args.batchSize ?? 50
const now = args.now ?? new Date()
const fetchImpl = args.fetchImpl ?? fetch
const pinnedFetchImpl = args.pinnedFetchImpl ?? pinnedHttpsFetch
const summary: DispatchSummary = { picked: 0, delivered: 0, failed: 0, dead: 0 }
@@ -138,7 +138,7 @@ export async function dispatchDueDeliveries(args: {
const outcome = await attemptDelivery({
delivery,
webhook,
fetchImpl,
pinnedFetchImpl,
now,
})
@@ -198,16 +198,12 @@ export async function dispatchDueDeliveries(args: {
*/
async function recoverStuckInFlight(supabase: SupabaseClient, now: Date): Promise<void> {
const stuckBefore = new Date(now.getTime() - 2 * REQUEST_TIMEOUT_MS)
// The status='in_flight' filter alone is not sufficient — a row could
// race between this SELECT and the UPDATE and reach 'delivered' or
// 'dead' in the interim. Postgres applies the status filter to the
// CURRENT (post-race) state, so the row would slip through and the
// immutability trigger would raise check_violation, aborting the
// entire bulk UPDATE and leaving legitimately stuck rows unrecovered.
//
// Defense-in-depth: explicitly exclude terminal status values. The
// partial guard makes a successful sweep on a mixed batch safe even
// when one row terminalized mid-flight.
// Under READ COMMITTED (Postgres default), UPDATE re-evaluates the WHERE
// clause against each row's current value when it acquires the row lock.
// A row that raced from 'in_flight' to 'delivered'/'dead' between scan
// and lock will fail status='in_flight' on re-evaluation and be skipped
// entirely — the immutability trigger never fires, so a mid-flight
// terminal flip cannot abort the bulk update.
const { data, error } = await supabase
.from('webhook_deliveries')
.update({
@@ -216,7 +212,6 @@ async function recoverStuckInFlight(supabase: SupabaseClient, now: Date): Promis
error: 'recovered_from_in_flight_timeout',
})
.eq('status', 'in_flight')
.not('status', 'in', '(delivered,dead)')
.lt('updated_at', stuckBefore.toISOString())
.select('id')
@@ -234,61 +229,26 @@ async function claimDueDeliveries(
batchSize: number,
now: Date,
): Promise<DueDelivery[]> {
// PostgREST cannot express FOR UPDATE SKIP LOCKED through the JS client.
// The cleaner long-term shape is a SQL claim function — tracked for a
// follow-up commit. Until then we SELECT candidate rows, then UPDATE
// with a CAS guard and `.select('id')` to learn which rows the UPDATE
// actually claimed. The dispatch loop runs ONLY against the intersection
// of (selected, claimed) — so an overlapping cron tick that picked up
// the same SELECT can never double-deliver: at most one tick wins the
// CAS update for any given row.
// Atomic FOR UPDATE SKIP LOCKED claim via the SQL function shipped in
// migration 20260515220000. PostgREST can't express SKIP LOCKED through
// the JS client, so the function form is the documented entry point —
// see the migration comment for the full rationale (one round trip,
// no CAS contention, rows locked by a concurrent tick are simply
// invisible to the second caller).
//
// Per-minute Vercel cron has best-effort single-instance semantics, but
// the documented contract is "at-least-once" not "at-most-once" — under
// load (e.g. a 50-row batch with mostly slow receivers > 60s) the next
// tick can fire while this one is still running, so the CAS-then-
// intersect pattern is load-bearing, not defensive.
const { data, error } = await supabase
.from('webhook_deliveries')
.select('id, webhook_id, company_id, event_type, payload, previous_attributes, api_version, attempts')
.in('status', ['pending', 'failed'])
.lte('next_attempt_at', now.toISOString())
// Skip dangling rows (webhook deleted between enqueue and dispatch).
// The webhook_deliveries.webhook_id FK is ON DELETE SET NULL
// (migration 20260515170000) so terminal rows survive webhook deletion
// for BFNAR 2013:2 kap 8 § audit retention; non-terminal rows for a
// deleted webhook have no receiver to deliver to and stay dormant in
// the audit trail.
.not('webhook_id', 'is', null)
.order('next_attempt_at', { ascending: true })
.limit(batchSize)
// All filter semantics from the previous JS path are preserved inside
// the function: status IN ('pending','failed'), next_attempt_at <= now,
// webhook_id IS NOT NULL, ORDER BY next_attempt_at ASC, LIMIT batchSize.
const { data, error } = await supabase.rpc('claim_due_webhook_deliveries', {
p_batch_size: batchSize,
p_now: now.toISOString(),
})
if (error || !data) {
log.error('claim due deliveries failed', error as Error)
if (error) {
log.error('claim_due_webhook_deliveries rpc failed', error as Error)
return []
}
if (data.length === 0) return []
const candidates = data as DueDelivery[]
const candidateIds = candidates.map((d) => d.id)
const { data: claimed, error: updateErr } = await supabase
.from('webhook_deliveries')
.update({ status: 'in_flight' })
.in('id', candidateIds)
.in('status', ['pending', 'failed']) // CAS guard
.select('id')
if (updateErr) {
log.error('claim deliveries update failed', updateErr as Error)
return []
}
// Trust the UPDATE's returned set as authoritative — anything not in
// `claimed` was lost to a competing tick (or had its status flipped
// out from under us between SELECT and UPDATE).
const claimedIds = new Set(((claimed ?? []) as { id: string }[]).map((r) => r.id))
return candidates.filter((d) => claimedIds.has(d.id))
return (data ?? []) as DueDelivery[]
}
async function loadWebhooksByIds(
@@ -439,10 +399,10 @@ type AttemptOutcome = DeliveredOutcome | FailedOutcome | DeadOutcome
async function attemptDelivery(args: {
delivery: DueDelivery
webhook: WebhookForDelivery
fetchImpl: typeof fetch
pinnedFetchImpl: typeof pinnedHttpsFetch
now: Date
}): Promise<AttemptOutcome> {
const { delivery, webhook, fetchImpl, now } = args
const { delivery, webhook, pinnedFetchImpl, now } = args
const attempts = delivery.attempts + 1
const requestId = `whdel_${delivery.id}`
@@ -455,135 +415,109 @@ async function attemptDelivery(args: {
previous_attributes: delivery.previous_attributes,
})
// Re-validate the URL at dispatch time as defense in depth — DNS records
// can change between webhook creation and dispatch (DNS rebinding,
// hijack, A-record swap to internal IP), so the create-time check alone
// is insufficient. A failure here marks the delivery dead with a
// distinct reason so the operator can investigate without thinking it's
// a transient receiver issue.
const urlCheck = await validateWebhookUrl(webhook.webhook_url)
if (!urlCheck.ok) {
return {
kind: 'dead',
reason: `url_unsafe:${urlCheck.reason}`,
disableWebhook: true,
attempts,
responseStatus: null,
responseBody: null,
responseHeaders: null,
error: urlCheck.detail,
}
}
const { header } = signPayload({
body,
secret: webhook.secret,
timestamp: Math.floor(now.getTime() / 1000),
})
const { header } = signPayload({ body, secret: webhook.secret, timestamp: Math.floor(now.getTime() / 1000) })
// pinnedHttpsFetch performs DNS validation AND opens the socket against
// the validated IP in a single call. The previous shape (separate
// validateWebhookUrl + fetch calls) left a DNS-rebinding window between
// the two — closed here. SNI + Host header continue to carry the
// original hostname so receiver-side TLS + vhost routing still work.
const result = await pinnedFetchImpl(webhook.webhook_url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Gnubok-Signature': header,
'X-Gnubok-Event': delivery.event_type,
'X-Gnubok-Delivery': delivery.id,
'X-Gnubok-Api-Version': delivery.api_version,
'X-Request-Id': requestId,
'User-Agent': 'gnubok-webhook/1',
},
body,
timeoutMs: REQUEST_TIMEOUT_MS,
maxResponseBytes: MAX_RESPONSE_BODY_BYTES,
})
const controller = new AbortController()
const timeout = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS)
let response: Response
try {
response = await fetchImpl(webhook.webhook_url, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-Gnubok-Signature': header,
'X-Gnubok-Event': delivery.event_type,
'X-Gnubok-Delivery': delivery.id,
'X-Gnubok-Api-Version': delivery.api_version,
'X-Request-Id': requestId,
'User-Agent': 'gnubok-webhook/1',
},
body,
signal: controller.signal,
// Reject 3xx responses entirely. Following a redirect would let a
// receiver bounce the dispatcher to a private/internal address
// AFTER the SSRF guard (which validated the original webhook_url's
// hostname) has cleared. Receivers that legitimately move endpoints
// should ask integrators to update the webhook URL.
redirect: 'error',
})
} catch (err) {
clearTimeout(timeout)
const message = err instanceof Error ? err.message : String(err)
// Distinguish redirect-rejection errors from generic transport
// failures. With redirect: 'error' the runtime fetch throws when the
// receiver returns 3xx — that's an SSRF-bypass attempt (or a
// misconfigured receiver), not a transient failure. Treating it as
// 'failed' would burn 8 retry attempts over ~72h before going dead.
// Mirror the HTTP 410 treatment: terminal + auto-disable so the
// operator surfaces the misbehaving receiver immediately.
//
// Node's undici (the runtime fetch) raises 'unexpected redirect'
// / 'redirect mode is set to error' messages; check both shapes
// since the exact wording has changed across Node versions.
const isRedirectError = /redirect/i.test(message)
if (isRedirectError) {
switch (result.kind) {
case 'unsafe_url':
return {
kind: 'dead',
reason: 'redirect_blocked',
reason: `url_unsafe:${result.reason}`,
disableWebhook: true,
attempts,
responseStatus: null,
responseBody: null,
responseHeaders: null,
error: message.length > 500 ? `${message.slice(0, 497)}...` : message,
error: result.detail,
}
case 'redirect_blocked':
return {
kind: 'dead',
reason: 'redirect_blocked',
disableWebhook: true,
attempts,
responseStatus: result.status,
responseBody: null,
responseHeaders: null,
error: truncateError(result.detail),
}
case 'timeout':
case 'transport_error':
return {
kind: 'failed',
attempts,
responseStatus: null,
responseBody: null,
responseHeaders: null,
error: truncateError(result.detail),
}
case 'ok': {
const responseHeaders = filterResponseHeaders(result.headers)
const responseBody = isSafeContentType(result.headers['content-type'] ?? '')
? result.body
: null
// HTTP 410 — receiver explicitly asks us to stop. Auto-disable.
if (result.status === 410) {
return {
kind: 'dead',
reason: 'http_410_gone',
disableWebhook: true,
attempts,
responseStatus: 410,
responseBody,
responseHeaders,
}
}
if (result.status >= 200 && result.status < 300) {
return {
kind: 'delivered',
attempts,
responseStatus: result.status,
responseBody,
responseHeaders,
}
}
return {
kind: 'failed',
attempts,
responseStatus: result.status,
responseBody,
responseHeaders,
error: `HTTP ${result.status}`,
}
}
return {
kind: 'failed',
attempts,
responseStatus: null,
responseBody: null,
responseHeaders: null,
error: message.length > 500 ? `${message.slice(0, 497)}...` : message,
}
}
}
// Keep the abort timeout armed across the body read — a slow body
// stream can stall the entire dispatch batch otherwise. Clear only
// after readBoundedText returns (or aborts).
let responseBody: string | null
try {
responseBody = await readBoundedText(response)
} finally {
clearTimeout(timeout)
}
const responseHeaders = headersToObject(response.headers)
// HTTP 410 — receiver explicitly asks us to stop. Auto-disable the
// webhook + mark this delivery dead.
if (response.status === 410) {
return {
kind: 'dead',
reason: 'http_410_gone',
disableWebhook: true,
attempts,
responseStatus: 410,
responseBody,
responseHeaders,
}
}
if (response.status >= 200 && response.status < 300) {
return {
kind: 'delivered',
attempts,
responseStatus: response.status,
responseBody,
responseHeaders,
}
}
return {
kind: 'failed',
attempts,
responseStatus: response.status,
responseBody,
responseHeaders,
error: `HTTP ${response.status}`,
}
function truncateError(message: string): string {
return message.length > 500 ? `${message.slice(0, 497)}...` : message
}
// Content-Type prefixes for which we persist response_body verbatim. Other
@@ -593,21 +527,9 @@ async function attemptDelivery(args: {
// when the operator can see the response_status and response_headers.
const SAFE_BODY_CONTENT_TYPE_PREFIXES = ['text/plain', 'application/json']
async function readBoundedText(response: Response): Promise<string | null> {
const contentType = response.headers.get('content-type')?.toLowerCase() ?? ''
const isSafe = SAFE_BODY_CONTENT_TYPE_PREFIXES.some((p) => contentType.startsWith(p))
if (!isSafe) {
// Drain the body so the connection can be reused, but discard the bytes.
try { await response.text() } catch { /* ignore */ }
return null
}
try {
const text = await response.text()
if (text.length <= MAX_RESPONSE_BODY_BYTES) return text
return text.slice(0, MAX_RESPONSE_BODY_BYTES)
} catch {
return null
}
function isSafeContentType(contentType: string): boolean {
const lower = contentType.toLowerCase()
return SAFE_BODY_CONTENT_TYPE_PREFIXES.some((p) => lower.startsWith(p))
}
// Allowlist for response_headers persistence. Receiver-side headers like
@@ -627,13 +549,13 @@ const SAFE_RESPONSE_HEADERS = new Set([
'cf-ray',
])
function headersToObject(headers: Headers): Record<string, string> {
function filterResponseHeaders(headers: Record<string, string>): Record<string, string> {
const obj: Record<string, string> = {}
headers.forEach((v, k) => {
for (const [k, v] of Object.entries(headers)) {
if (SAFE_RESPONSE_HEADERS.has(k.toLowerCase())) {
obj[k] = v
}
})
}
return obj
}
lib/webhooks/pinned-fetch.ts
+275
View File
@@ -0,0 +1,275 @@
/**
* Pinned-IP HTTPS POST for webhook dispatch.
*
* Closes the DNS-rebinding window between url-guard validation and the
* actual HTTPS request. The previous shape was:
*
* 1. validateWebhookUrl() → DNS resolves to [public IP], returns ok
* 2. fetch(webhook_url) → re-resolves DNS; an attacker who flipped
* the A record in the interval gets a
* private-IP socket
*
* The new shape pins the request to the IP validated in step 1, with the
* original hostname carried in:
* - the TLS SNI extension (so the receiver's cert continues to match)
* - the HTTP Host header (so vhost routing on the receiver continues to
* work)
*
* The request socket therefore never re-resolves DNS, foreclosing the
* rebind race. Documented openly per the url-guard.ts file header
* ("closing that requires a custom HTTPS agent that pins the resolved IP").
*
* Built on `node:https.request` rather than undici's Agent because (a) the
* project doesn't take a dependency on undici, (b) the stdlib API is more
* explicit about the SNI / Host / IP split, (c) https.request is enough
* for HTTP/1.1 + TLS, which every webhook receiver supports.
*
* Inversion seam: `httpsRequest` injectable for tests so we don't need to
* stand up an HTTPS server to verify the pinning / SNI / Host shape. The
* dispatcher's tests pass a stub through `pinnedFetchImpl`.
*/
import {
request as httpsRequestDefault,
type RequestOptions as HttpsRequestOptions,
} from 'node:https'
import type { ClientRequest, IncomingMessage } from 'node:http'
import { validateWebhookUrl as validateWebhookUrlDefault } from './url-guard'
export type PinnedFetchResult =
| {
kind: 'ok'
status: number
headers: Record<string, string>
body: string
bodyTruncated: boolean
pinnedAddress: string
}
| { kind: 'unsafe_url'; reason: string; detail: string; pinnedAddress: null }
| { kind: 'redirect_blocked'; status: number; detail: string; pinnedAddress: string }
| { kind: 'timeout'; detail: string; pinnedAddress: string }
| { kind: 'transport_error'; detail: string; pinnedAddress: string | null }
export interface PinnedFetchInit {
method: string
headers: Record<string, string>
body: string
timeoutMs: number
/** Max bytes captured from response body — receivers returning long error pages get truncated. */
maxResponseBytes: number
}
export interface PinnedFetchDeps {
/** DNS validation seam. Defaults to url-guard's validateWebhookUrl. */
validateUrl?: typeof validateWebhookUrlDefault
/** Raw HTTPS request seam. Defaults to node:https.request. */
httpsRequest?: (
options: HttpsRequestOptions,
callback: (res: IncomingMessage) => void,
) => ClientRequest
}
export async function pinnedHttpsFetch(
rawUrl: string,
init: PinnedFetchInit,
deps: PinnedFetchDeps = {},
): Promise<PinnedFetchResult> {
const validateUrl = deps.validateUrl ?? validateWebhookUrlDefault
const httpsRequest = deps.httpsRequest ?? httpsRequestDefault
let parsed: URL
try {
parsed = new URL(rawUrl)
} catch {
return {
kind: 'unsafe_url',
reason: 'invalid_url',
detail: 'URL did not parse.',
pinnedAddress: null,
}
}
const validation = await validateUrl(rawUrl)
if (!validation.ok) {
return {
kind: 'unsafe_url',
reason: validation.reason,
detail: validation.detail,
pinnedAddress: null,
}
}
// Pick the first vetted address. validateWebhookUrl rejects the whole
// set when ANY entry is unsafe, so the first is safe by construction.
// Deterministic choice keeps log output stable across retries.
const pinnedAddress = validation.resolvedAddresses[0]
if (!pinnedAddress) {
// Defensive — validateWebhookUrl returns ok only when there's at least
// one address, but a future refactor could regress this and we want
// the failure to be loud, not a silent DNS-lookup-by-empty-host.
return {
kind: 'transport_error',
detail: 'No resolved address from validateWebhookUrl',
pinnedAddress: null,
}
}
const port = parsed.port ? Number(parsed.port) : 443
return new Promise<PinnedFetchResult>((resolve) => {
let settled = false
const settle = (r: PinnedFetchResult) => {
if (settled) return
settled = true
resolve(r)
}
// The HTTP Host header must carry the original hostname (vhost routing
// on the receiver). Include the port only when non-default — RFC 7230
// §5.4 says the port is omitted when it matches the scheme default.
const hostHeader = port === 443 ? parsed.hostname : `${parsed.hostname}:${port}`
const requestOptions: HttpsRequestOptions = {
protocol: 'https:',
// Pin the socket to the validated IP. node:https accepts the
// address directly — no further DNS lookup happens.
host: pinnedAddress,
port,
path: parsed.pathname + parsed.search,
method: init.method,
// SNI carries the original hostname so the receiver's TLS cert
// (which is issued for the hostname, not the IP) validates.
//
// Cert-vs-hostname verification: Node's default checkServerIdentity
// matches the cert's SAN/CN against `servername` (or `host` when
// servername is unset). Because `servername` is set to the original
// hostname, the IP substitution above does NOT weaken the hostname-
// verification step — a forged endpoint at the pinned IP presenting
// a valid cert for a DIFFERENT hostname would fail the handshake.
// No explicit checkServerIdentity override is needed; relying on
// the default is the documented contract.
servername: parsed.hostname,
headers: {
...init.headers,
// Lowercase 'host' — Node's https.request would synthesise one
// from `host` (the pinned IP) if we didn't set it explicitly,
// which would break vhost routing on the receiver.
host: hostHeader,
},
// Fresh socket per call — webhook delivery doesn't benefit from
// Keep-Alive (the dispatcher serializes and the IP changes per
// dispatch from re-validation). agent:false also forecloses any
// accidental pool-level reuse across pinned IPs.
agent: false,
}
let absoluteTimer: NodeJS.Timeout | null = null
const req = httpsRequest(requestOptions, (res) => {
// Receivers MUST return a non-redirect. Following a 3xx would let
// them bounce the dispatcher to a private address AFTER the SSRF
// guard cleared. We don't follow redirects; treat as terminal here
// and let the dispatcher mark the row dead with reason='redirect_
// blocked' for consistency with the old fetch path's behavior.
const status = res.statusCode ?? 0
if (status >= 300 && status < 400) {
// Drain body so the socket cleans up; ignore errors.
res.resume()
req.destroy()
if (absoluteTimer) clearTimeout(absoluteTimer)
return settle({
kind: 'redirect_blocked',
status,
detail: `Receiver returned ${status}; redirects are refused.`,
pinnedAddress,
})
}
const chunks: Buffer[] = []
let total = 0
let truncated = false
res.on('data', (chunk: Buffer) => {
if (truncated) return
if (total + chunk.length > init.maxResponseBytes) {
const remaining = init.maxResponseBytes - total
if (remaining > 0) chunks.push(chunk.subarray(0, remaining))
total = init.maxResponseBytes
truncated = true
// Destroy the stream — no point pulling the rest over the wire.
res.destroy()
} else {
chunks.push(chunk)
total += chunk.length
}
})
const finalize = () => {
if (absoluteTimer) clearTimeout(absoluteTimer)
const headers: Record<string, string> = {}
for (const [k, v] of Object.entries(res.headers)) {
if (typeof v === 'string') headers[k] = v
else if (Array.isArray(v)) headers[k] = v.join(', ')
}
settle({
kind: 'ok',
status,
headers,
body: Buffer.concat(chunks).toString('utf8'),
bodyTruncated: truncated,
pinnedAddress,
})
}
// Two completion paths to handle: 'end' (normal completion) and
// 'close' (when we destroyed the stream for size truncation, where
// 'end' does not fire). Node emits BOTH 'end' and 'close' on normal
// completions, so `once()` + a self-removing pair keeps finalize
// single-shot without relying on the outer `settled` guard to
// squash duplicate header reconstruction.
const finalizeOnce = () => {
res.removeListener('end', finalizeOnce)
res.removeListener('close', finalizeOnce)
finalize()
}
res.once('end', finalizeOnce)
res.once('close', finalizeOnce)
res.on('error', (err) => {
if (absoluteTimer) clearTimeout(absoluteTimer)
settle({ kind: 'transport_error', detail: err.message, pinnedAddress })
})
})
// Two-layer timeout: socket-idle timeout via Node's built-in, plus a
// wall-clock absolute timeout. node:https `timeout` is idle-only and
// wouldn't fire if a slow receiver dribbles bytes; the absolute timer
// is the hard cap.
req.setTimeout(init.timeoutMs)
req.on('timeout', () => {
req.destroy()
if (absoluteTimer) clearTimeout(absoluteTimer)
settle({
kind: 'timeout',
detail: `Socket idle for ${init.timeoutMs} ms`,
pinnedAddress,
})
})
absoluteTimer = setTimeout(() => {
req.destroy()
settle({
kind: 'timeout',
detail: `Request exceeded ${init.timeoutMs} ms wall-clock`,
pinnedAddress,
})
}, init.timeoutMs)
req.on('error', (err) => {
if (absoluteTimer) clearTimeout(absoluteTimer)
settle({ kind: 'transport_error', detail: err.message, pinnedAddress })
})
if (init.body) req.write(init.body)
req.end()
})
}
supabase/migrations/20260515210000_operations_immutability.sql
+78
View File
@@ -0,0 +1,78 @@
-- Migration: operations_immutability
--
-- BFNAR 2013:2 kap 8 § behandlingshistorik integrity: an audit row that
-- records the outcome of a system event becomes immutable once finalised.
-- For the v1 `operations` table the terminal states are `succeeded`,
-- `failed`, and `cancelled` — once any of those is set, the row records
-- what happened and must not be re-mutated. A future bug, a privileged
-- operator, or a compromised service-role caller cannot rewrite "this
-- year-end close succeeded" to "failed".
--
-- This mirrors the trigger pair the webhook_deliveries table got in
-- 20260515170000 (BEFORE UPDATE) + 20260515190000 (BEFORE DELETE). Same
-- predicate shape, same ERRCODE = check_violation, same SECURITY DEFINER
-- + search_path = public.
--
-- The lifecycle helpers (lib/api/v1/operations.ts) continue to legitimately
-- transition `running → succeeded/failed/cancelled` because OLD.status is
-- `running` (non-terminal) at the moment the UPDATE fires. Only post-
-- terminal mutations are blocked.
--
-- Carry-over from Phase 4 PR-2 (PR #469) review rounds: Swedish-compliance
-- flagged that the operations table allowed UPDATE/DELETE of result / error
-- / status on rows already in terminal status. Deferred at the time; closed
-- here as part of the Phase 6 PR-3 substrate-hardening pass.
-- ──────────────────────────────────────────────────────────────────────
-- 1. BEFORE UPDATE — block mutations to terminal-status rows
-- ──────────────────────────────────────────────────────────────────────
CREATE OR REPLACE FUNCTION public.enforce_operation_immutability()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public
AS $$
BEGIN
IF OLD.status IN ('succeeded', 'failed', 'cancelled') THEN
RAISE EXCEPTION 'operations row in terminal status (%) is immutable', OLD.status
USING ERRCODE = 'check_violation';
END IF;
RETURN NEW;
END;
$$;
-- Name starts with 'e' so it fires before `operations_updated_at` (starts
-- with 'o') in default alphabetical trigger order. Belt-and-braces — the
-- RAISE would abort the entire UPDATE regardless, but firing first keeps
-- the failed-write trail tidy in pg_stat_user_tables.
CREATE TRIGGER enforce_operation_immutability
BEFORE UPDATE ON public.operations
FOR EACH ROW EXECUTE FUNCTION public.enforce_operation_immutability();
-- ──────────────────────────────────────────────────────────────────────
-- 2. BEFORE DELETE — block hard-delete of terminal-status rows
-- ──────────────────────────────────────────────────────────────────────
CREATE OR REPLACE FUNCTION public.block_operation_terminal_delete()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public
AS $$
BEGIN
IF OLD.status IN ('succeeded', 'failed', 'cancelled') THEN
RAISE EXCEPTION
'operations row in terminal status (%) cannot be deleted (BFNAR 2013:2 kap 8 § behandlingshistorik integrity)',
OLD.status
USING ERRCODE = 'check_violation';
END IF;
RETURN OLD;
END;
$$;
CREATE TRIGGER block_operation_terminal_delete
BEFORE DELETE ON public.operations
FOR EACH ROW EXECUTE FUNCTION public.block_operation_terminal_delete();
NOTIFY pgrst, 'reload schema';
supabase/migrations/20260515220000_claim_due_webhook_deliveries.sql
+100
View File
@@ -0,0 +1,100 @@
-- Migration: claim_due_webhook_deliveries
--
-- Replaces the dispatcher's SELECT-then-UPDATE intersect pattern with a
-- single atomic SQL function using FOR UPDATE SKIP LOCKED. PostgREST cannot
-- express SKIP LOCKED through the JS client, so the previous shape (read
-- lib/webhooks/dispatcher.ts:232–292) did:
--
-- 1. SELECT pending/failed rows ordered by next_attempt_at
-- 2. UPDATE WHERE id IN (...) AND status IN ('pending','failed') -- CAS
-- 3. Intersect (selected, returned-from-UPDATE) → claim set
--
-- That pattern is correct under concurrent ticks — the CAS guard ensures
-- only one tick wins per row — but burns two round trips per cycle and
-- doesn't communicate the locking semantics. Under load (slow receivers
-- stretching a tick past 60 s while the next minute's cron starts) both
-- ticks have to negotiate which rows they actually own.
--
-- The function form uses SKIP LOCKED inside a CTE, so a row already locked
-- by a concurrent tick is simply invisible to the second caller — no CAS
-- contention, one round trip. The dispatch loop becomes:
--
-- const { data } = await supabase.rpc('claim_due_webhook_deliveries', {
-- p_batch_size: 50, p_now: new Date().toISOString(),
-- })
--
-- All filter semantics from the existing JS path are preserved:
-- - status IN ('pending','failed') (non-terminal, due-able)
-- - next_attempt_at <= p_now (genuinely due)
-- - webhook_id IS NOT NULL (dangling rows go dormant
-- under the FK SET NULL
-- from 20260515170000)
-- - ORDER BY next_attempt_at ASC (oldest-due-first)
-- - LIMIT p_batch_size (back-pressure)
--
-- The immutability trigger (enforce_webhook_delivery_immutability) is
-- already correct for this path: it RAISES when OLD.status is terminal
-- ('delivered', 'dead'); rows here have OLD.status in ('pending', 'failed')
-- so the UPDATE passes.
--
-- SECURITY DEFINER because the dispatcher runs under createServiceClient-
-- NoCookies (service-role), and a future change that hardens RLS or
-- restricts the service_role's UPDATE access on webhook_deliveries should
-- not silently break dispatch. The function is the documented entry point
-- for the dispatcher loop.
CREATE OR REPLACE FUNCTION public.claim_due_webhook_deliveries(
p_batch_size int,
p_now timestamptz DEFAULT now()
)
RETURNS TABLE (
id uuid,
webhook_id uuid,
company_id uuid,
event_type text,
payload jsonb,
previous_attributes jsonb,
api_version text,
attempts int
)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public
AS $$
BEGIN
-- Reject obviously-bad batch sizes early. A negative or zero batch size
-- would degenerate the CTE into a no-op; a runaway value (e.g. an
-- accidentally unbounded query) could lock too many rows in one tick
-- and starve the next.
IF p_batch_size IS NULL OR p_batch_size <= 0 OR p_batch_size > 1000 THEN
RAISE EXCEPTION 'p_batch_size must be in (0, 1000]; got %', p_batch_size
USING ERRCODE = 'invalid_parameter_value';
END IF;
RETURN QUERY
WITH due AS (
SELECT wd.id
FROM public.webhook_deliveries wd
WHERE wd.status IN ('pending', 'failed')
AND wd.next_attempt_at <= p_now
AND wd.webhook_id IS NOT NULL
ORDER BY wd.next_attempt_at ASC
LIMIT p_batch_size
FOR UPDATE SKIP LOCKED
)
UPDATE public.webhook_deliveries wd
SET status = 'in_flight'
FROM due
WHERE wd.id = due.id
RETURNING wd.id,
wd.webhook_id,
wd.company_id,
wd.event_type,
wd.payload,
wd.previous_attributes,
wd.api_version,
wd.attempts;
END;
$$;
NOTIFY pgrst, 'reload schema';
supabase/migrations/20260515230000_webhook_deliveries_body_length_check.sql
+21
View File
@@ -0,0 +1,21 @@
-- Migration: webhook_deliveries_body_length_check
--
-- Defense-in-depth size cap on webhook_deliveries.response_body. The
-- application layer already truncates response bodies to 4 KB in
-- lib/webhooks/pinned-fetch.ts (the MAX_RESPONSE_BODY_BYTES constant
-- imported from the dispatcher). A future refactor that accidentally
-- bypasses that truncation — or a non-dispatcher write path that lands
-- in this column — would silently store large blobs in a column that
-- sits next to event payloads carrying personal data.
--
-- Adds a hard ceiling at the DB layer so any path that tries to write
-- a longer string is surfaced as a check_violation rather than persisted
-- (Art.32(1)(b) integrity of processing, A.8.24 protection of records).
-- The ceiling is set generously above the application limit so a
-- legitimate dispatcher write never hits this — only a regression would.
ALTER TABLE public.webhook_deliveries
ADD CONSTRAINT webhook_deliveries_response_body_length_check
CHECK (response_body IS NULL OR length(response_body) <= 8192);
NOTIFY pgrst, 'reload schema';