diff --git a/apps/lifecycle/src/campaign/send.spec.ts b/apps/lifecycle/src/campaign/send.spec.ts index 7d86163b5..c254b5e63 100644 --- a/apps/lifecycle/src/campaign/send.spec.ts +++ b/apps/lifecycle/src/campaign/send.spec.ts @@ -504,6 +504,7 @@ describe('dispatchLifecycleAppOwnedJob', () => { 'campaign_disabled', 'delivery_disabled', 'outside_send_window', + 'reply_binding_pending', ] as const)( 'keeps an install-runtime hello deferred while %s', async (reason) => { diff --git a/apps/lifecycle/src/campaign/send.ts b/apps/lifecycle/src/campaign/send.ts index a92d2a183..6a0e3caf3 100644 --- a/apps/lifecycle/src/campaign/send.ts +++ b/apps/lifecycle/src/campaign/send.ts @@ -404,7 +404,8 @@ async function dispatchRecipient( if ( result.reason === 'campaign_disabled' || result.reason === 'delivery_disabled' || - result.reason === 'outside_send_window' + result.reason === 'outside_send_window' || + result.reason === 'reply_binding_pending' ) { const now = dependencies.now(); await dependencies.deferJob(executor, { diff --git a/apps/lifecycle/src/dispatcher.spec.ts b/apps/lifecycle/src/dispatcher.spec.ts index 73f7ebed4..fd1fd32e9 100644 --- a/apps/lifecycle/src/dispatcher.spec.ts +++ b/apps/lifecycle/src/dispatcher.spec.ts @@ -1,6 +1,7 @@ import { createUnsubscribeActionUrl, dispatchGrowthLeasedJob, + reconcilePendingResendMessageIds, type GrowthJob, type SqlExecutor, } from '@threadplane-internal/growth'; @@ -63,6 +64,7 @@ function dependencies( createDatabase: vi.fn(() => executor), dispatchLeasedJob: vi.fn().mockResolvedValue('completed'), isRecoveryPaused: vi.fn().mockResolvedValue(false), + reconcileMessageIds: vi.fn().mockResolvedValue({ attempted: 0, bound: 0 }), leaseDueJobs: vi.fn().mockResolvedValue([]), loadEmailKeyring: vi.fn(() => EMAIL_KEYRING), processInstallRuntimeActivations: vi.fn().mockResolvedValue({ @@ -772,3 +774,50 @@ describe('dispatchLifecycleJobs', () => { expect(dispatchLeasedJob).toHaveBeenCalledTimes(2); }); }); + +it('reconciles accepted send identities before leasing contact follow-ups', async () => { + const deps = dependencies(); + await dispatchLifecycleJobs( + { + batchSize: 5, + campaignEnabled: true, + signal: new AbortController().signal, + }, + deps + ); + expect(deps.reconcileMessageIds).toHaveBeenCalledWith(expect.anything(), { + now: NOW, + signal: expect.any(AbortSignal), + }); + expect( + deps.reconcileMessageIds && + vi.mocked(deps.reconcileMessageIds).mock.invocationCallOrder[0] + ).toBeLessThan(vi.mocked(deps.leaseDueJobs).mock.invocationCallOrder[0] ?? 0); +}); + +it('continues enrichment-only dispatch when delivery configuration is absent', async () => { + vi.stubEnv('RESEND_API_KEY', undefined); + vi.stubEnv('DELIVERY_ENVIRONMENT', undefined); + vi.stubEnv('GROWTH_DATABASE_ENVIRONMENT', undefined); + try { + const deps = dependencies({ + reconcileMessageIds: reconcilePendingResendMessageIds, + leaseDueJobs: vi + .fn() + .mockResolvedValue([leasedJob('enrich-1', 'enrich')]), + }); + await expect( + dispatchLifecycleJobs( + { + batchSize: 5, + campaignEnabled: true, + signal: new AbortController().signal, + }, + deps + ) + ).resolves.toMatchObject({ dispatched: 1 }); + expect(deps.dispatchLeasedJob).toHaveBeenCalledOnce(); + } finally { + vi.unstubAllEnvs(); + } +}); diff --git a/apps/lifecycle/src/dispatcher.ts b/apps/lifecycle/src/dispatcher.ts index c9ab4295d..5d8f6f487 100644 --- a/apps/lifecycle/src/dispatcher.ts +++ b/apps/lifecycle/src/dispatcher.ts @@ -6,6 +6,7 @@ import { leaseDueJobs, materializeCampaignEnrollment, processInstallRuntimeActivations, + reconcilePendingResendMessageIds, renewJobLease, type GrowthAppJobHandlers, type GrowthDispatchDependencies, @@ -60,6 +61,7 @@ export interface LifecycleDispatcherDependencies { leaseDueJobs: typeof leaseDueJobs; materializeCampaignEnrollment: typeof materializeCampaignEnrollment; processInstallRuntimeActivations: typeof processInstallRuntimeActivations; + reconcileMessageIds?: typeof reconcilePendingResendMessageIds; loadEmailKeyring: typeof loadEmailHmacKeyring; now: () => Date; renewJobLease: typeof renewJobLease; @@ -76,6 +78,7 @@ const defaultDependencies: LifecycleDispatcherDependencies = { leaseDueJobs, materializeCampaignEnrollment, processInstallRuntimeActivations, + reconcileMessageIds: reconcilePendingResendMessageIds, loadEmailKeyring: loadEmailHmacKeyring, now: () => new Date(), renewJobLease, @@ -157,6 +160,11 @@ export async function dispatchLifecycleJobs( input.signal.throwIfAborted(); const executor = dependencies.createDatabase(); try { + await dependencies.reconcileMessageIds?.(executor, { + now: dependencies.now(), + signal: input.signal, + }); + input.signal.throwIfAborted(); if (input.campaignEnrollmentEnabled) { if ( !(input.campaignEnrollmentStartAt instanceof Date) || diff --git a/apps/website/src/app/api/leads/route.ts b/apps/website/src/app/api/leads/route.ts index 5f4666c5e..e8d37a315 100644 --- a/apps/website/src/app/api/leads/route.ts +++ b/apps/website/src/app/api/leads/route.ts @@ -8,6 +8,8 @@ import { import { matchesSubmittedFormPolicy } from '../../../lib/growth/form-policy'; import { defaultGrowthFormRouteDependencies, + formAdmissionError, + trustedFormClientIp, jsonResponse, readBoundedJsonObject, stalePolicyResponse, @@ -57,6 +59,7 @@ export function createLeadRoute( return jsonResponse({ error: 'Invalid form' }, 400); } + let honeypot; let submissionId; let acquisitionSessionId; let email; @@ -67,6 +70,7 @@ export function createLeadRoute( let timeline; let pilotInterest; try { + honeypot = strictText(body, 'website_url', 200); submissionId = strictText(body, 'submission_id', 36); acquisitionSessionId = strictText(body, 'acquisition_session_id', 36); email = strictText(body, 'email', 254); @@ -115,8 +119,13 @@ export function createLeadRoute( } let accepted = false; + const trustedClientIp = trustedFormClientIp(request); + let deliverySuppressed = false; + let admissionError: Response | undefined; try { - await dependencies.accept(database, { + const result = await dependencies.accept(database, { + ...(honeypot ? { honeypot } : {}), + ...(trustedClientIp ? { trustedClientIp } : {}), submissionId, email: normalizedEmail, displayName: name || undefined, @@ -132,7 +141,9 @@ export function createLeadRoute( keyring, }); accepted = true; - } catch { + deliverySuppressed = result.deliverySuppressed === true; + } catch (error) { + admissionError = formAdmissionError(error); // The response below reports the failure without echoing provider detail. } @@ -141,7 +152,9 @@ export function createLeadRoute( } catch { return unableToAccept(); } + if (admissionError) return admissionError; if (!accepted) return unableToAccept(); + if (deliverySuppressed) return jsonResponse({ ok: true }); // The durable jobs remain available to the scheduled dispatcher. await dependencies.nudge({ submissionId }).catch(() => undefined); diff --git a/apps/website/src/app/api/newsletter/route.ts b/apps/website/src/app/api/newsletter/route.ts index 52ffd4424..0eff5ea7c 100644 --- a/apps/website/src/app/api/newsletter/route.ts +++ b/apps/website/src/app/api/newsletter/route.ts @@ -5,6 +5,8 @@ import { normalizeRecipientEmail } from '@threadplane-internal/growth'; import { matchesSubmittedFormPolicy } from '../../../lib/growth/form-policy'; import { defaultGrowthFormRouteDependencies, + formAdmissionError, + trustedFormClientIp, jsonResponse, readBoundedJsonObject, stalePolicyResponse, @@ -30,6 +32,7 @@ export function createNewsletterRoute( return jsonResponse({ error: 'Unable to accept request' }, 503); } + let honeypot; let submissionId; let acquisitionSessionId; let email; @@ -38,6 +41,7 @@ export function createNewsletterRoute( if (!matchesSubmittedFormPolicy(policy, policyVersion || undefined)) { return stalePolicyResponse(policy); } + honeypot = strictText(body, 'website_url', 200); submissionId = strictText(body, 'submission_id', 36); acquisitionSessionId = strictText(body, 'acquisition_session_id', 36); email = strictText(body, 'email', 254); @@ -65,8 +69,13 @@ export function createNewsletterRoute( } let accepted = false; + const trustedClientIp = trustedFormClientIp(request); + let deliverySuppressed = false; + let admissionError: Response | undefined; try { - await dependencies.accept(database, { + const result = await dependencies.accept(database, { + ...(honeypot ? { honeypot } : {}), + ...(trustedClientIp ? { trustedClientIp } : {}), submissionId, email: normalizedEmail, form: { kind: 'newsletter' }, @@ -80,7 +89,9 @@ export function createNewsletterRoute( keyring, }); accepted = true; - } catch { + deliverySuppressed = result.deliverySuppressed === true; + } catch (error) { + admissionError = formAdmissionError(error); // The response below reports the failure without echoing provider detail. } @@ -89,7 +100,9 @@ export function createNewsletterRoute( } catch { return unableToAccept(); } + if (admissionError) return admissionError; if (!accepted) return unableToAccept(); + if (deliverySuppressed) return jsonResponse({ ok: true }); // The durable jobs remain available to the scheduled dispatcher. await dependencies.nudge({ submissionId }).catch(() => undefined); diff --git a/apps/website/src/app/api/webhooks/resend/route.spec.ts b/apps/website/src/app/api/webhooks/resend/route.spec.ts index be4f1010b..0bfae23a8 100644 --- a/apps/website/src/app/api/webhooks/resend/route.spec.ts +++ b/apps/website/src/app/api/webhooks/resend/route.spec.ts @@ -69,7 +69,10 @@ describe('/api/webhooks/resend', () => { const body = rawPayload.replace('email.delivered', 'email.sent'); test.verify.mockImplementationOnce(() => { test.order.push('verify'); - return { type: 'email.sent', data: { email_id: 'resend-email-1' } }; + return { + type: 'email.sent', + data: { email_id: 'resend-email-1', message_id: '' }, + }; }); const response = await test.POST(request(body) as never); @@ -90,7 +93,13 @@ describe('/api/webhooks/resend', () => { test.database, { providerEventId: 'msg_123', - payload: { type: 'email.sent', data: { email_id: 'resend-email-1' } }, + payload: { + type: 'email.sent', + data: { + email_id: 'resend-email-1', + message_id: '', + }, + }, } ); expect(test.database.close).toHaveBeenCalledTimes(1); @@ -215,7 +224,9 @@ describe('/api/webhooks/resend', () => { const second = await test.POST(request() as never); const third = await test.POST(request() as never); - expect([first.status, second.status, third.status]).toEqual([503, 200, 200]); + expect([first.status, second.status, third.status]).toEqual([ + 503, 200, 200, + ]); expect(test.processVerifiedResendWebhook).toHaveBeenCalledTimes(3); expect(test.createDatabase).toHaveBeenCalledTimes(3); expect(test.database.close).toHaveBeenCalledTimes(3); diff --git a/apps/website/src/app/api/whitepaper-signup/route.ts b/apps/website/src/app/api/whitepaper-signup/route.ts index cedbad354..695af1a21 100644 --- a/apps/website/src/app/api/whitepaper-signup/route.ts +++ b/apps/website/src/app/api/whitepaper-signup/route.ts @@ -5,6 +5,8 @@ import { normalizeRecipientEmail } from '@threadplane-internal/growth'; import { matchesSubmittedFormPolicy } from '../../../lib/growth/form-policy'; import { defaultGrowthFormRouteDependencies, + formAdmissionError, + trustedFormClientIp, jsonResponse, readBoundedJsonObject, stalePolicyResponse, @@ -39,6 +41,7 @@ export function createWhitepaperSignupRoute( return jsonResponse({ error: 'Unable to accept request' }, 503); } + let honeypot; let submissionId; let acquisitionSessionId; let name; @@ -49,6 +52,7 @@ export function createWhitepaperSignupRoute( if (!matchesSubmittedFormPolicy(policy, policyVersion || undefined)) { return stalePolicyResponse(policy); } + honeypot = strictText(body, 'website_url', 200); submissionId = strictText(body, 'submission_id', 36); acquisitionSessionId = strictText(body, 'acquisition_session_id', 36); name = strictText(body, 'name', 200); @@ -81,8 +85,13 @@ export function createWhitepaperSignupRoute( } let accepted = false; + const trustedClientIp = trustedFormClientIp(request); + let deliverySuppressed = false; + let admissionError: Response | undefined; try { - await dependencies.accept(database, { + const result = await dependencies.accept(database, { + ...(honeypot ? { honeypot } : {}), + ...(trustedClientIp ? { trustedClientIp } : {}), submissionId, email: normalizedEmail, displayName: name || undefined, @@ -97,7 +106,9 @@ export function createWhitepaperSignupRoute( keyring, }); accepted = true; - } catch { + deliverySuppressed = result.deliverySuppressed === true; + } catch (error) { + admissionError = formAdmissionError(error); // The response below reports the failure without echoing provider detail. } @@ -106,7 +117,9 @@ export function createWhitepaperSignupRoute( } catch { return unableToAccept(); } + if (admissionError) return admissionError; if (!accepted) return unableToAccept(); + if (deliverySuppressed) return jsonResponse({ ok: true }); // The durable jobs remain available to the scheduled dispatcher. await dependencies.nudge({ submissionId }).catch(() => undefined); diff --git a/apps/website/src/components/contact/ContactForm.tsx b/apps/website/src/components/contact/ContactForm.tsx index aadecab1c..bd28a51c3 100644 --- a/apps/website/src/components/contact/ContactForm.tsx +++ b/apps/website/src/components/contact/ContactForm.tsx @@ -1,4 +1,5 @@ 'use client'; +import { Honeypot, readHoneypot } from '../form/Honeypot'; import React, { useState } from 'react'; import { Button } from '../ui/Button'; @@ -88,6 +89,7 @@ export function ContactForm({ formPolicy, intent = 'contact', entryPoint }: Cont return; } void form.submit({ + website_url: readHoneypot(e.currentTarget), form_kind: enterprise ? 'pricing' : 'contact', email: email.trim(), ...(name.trim() ? { name: name.trim() } : {}), @@ -113,6 +115,7 @@ export function ContactForm({ formPolicy, intent = 'contact', entryPoint }: Cont return (
+ { + const { container } = render( + + + + ); + const input = container.querySelector('input'); + const form = container.querySelector('form'); + if (!input || !form) throw new Error('Expected form trap'); + expect(input.tabIndex).toBe(-1); + expect(input.getAttribute('autocomplete')).toBe('off'); + expect(input.closest('[aria-hidden="true"]')).not.toBeNull(); + expect(readHoneypot(form)).toBe(''); + fireEvent.change(input, { target: { value: 'https://spam.invalid' } }); + expect(readHoneypot(form)).toBe( + 'https://spam.invalid' + ); +}); diff --git a/apps/website/src/components/form/Honeypot.tsx b/apps/website/src/components/form/Honeypot.tsx new file mode 100644 index 000000000..9ee4a8364 --- /dev/null +++ b/apps/website/src/components/form/Honeypot.tsx @@ -0,0 +1,24 @@ +'use client'; + +/** A bot trap: intentionally absent from keyboard and assistive navigation. */ +export function Honeypot() { + return ( + + ); +} + +export function readHoneypot(form: HTMLFormElement): string { + const value = new FormData(form).get('website_url'); + return typeof value === 'string' ? value : ''; +} diff --git a/apps/website/src/components/form/index.ts b/apps/website/src/components/form/index.ts index 47bcaf0ff..9197d82a1 100644 --- a/apps/website/src/components/form/index.ts +++ b/apps/website/src/components/form/index.ts @@ -11,3 +11,4 @@ export type { UseGrowthFormOptions, GrowthFormRoute, } from './use-growth-form'; +export { Honeypot, readHoneypot } from './Honeypot'; diff --git a/apps/website/src/components/landing/WhitePaperForm.tsx b/apps/website/src/components/landing/WhitePaperForm.tsx index 15b49d0df..18313eb40 100644 --- a/apps/website/src/components/landing/WhitePaperForm.tsx +++ b/apps/website/src/components/landing/WhitePaperForm.tsx @@ -1,4 +1,5 @@ 'use client'; +import { Honeypot, readHoneypot } from '../form/Honeypot'; import { useState } from 'react'; import type { FormEvent } from 'react'; import type { PublicFormPolicy } from '../../lib/growth/form-policy'; @@ -42,7 +43,7 @@ export function WhitePaperForm({ const pdf = PDF_PATHS[paper]; const [email, setEmail] = useState(''); const [emailMessage, setEmailMessage] = useState(null); - const form = useGrowthForm<{ email: string; paper: WhitepaperId }>({ + const form = useGrowthForm<{ website_url: string; email: string; paper: WhitepaperId }>({ route: '/api/whitepaper-signup', formPolicy, events: { @@ -71,7 +72,7 @@ export function WhitePaperForm({ ); - const submit = (e: FormEvent) => { + const submit = (e: FormEvent) => { e.preventDefault(); const problem = emailError(email); setEmailMessage(problem); @@ -79,7 +80,7 @@ export function WhitePaperForm({ document.getElementById(inputId)?.focus(); return; } - void form.submit({ email: email.trim(), paper }); + void form.submit({ website_url: readHoneypot(e.currentTarget), email: email.trim(), paper }); }; if (form.status === 'sent') { @@ -100,6 +101,7 @@ export function WhitePaperForm({ } return (
+
(null); const disclosureId = 'toast-whitepaper-growth-disclosure'; - const form = useGrowthForm<{ email: string; paper: 'overview' }>({ + const form = useGrowthForm<{ website_url: string; email: string; paper: 'overview' }>({ route: '/api/whitepaper-signup', formPolicy, events: { @@ -123,7 +124,7 @@ export function AnnouncementToast({ // dismiss is stable for the component's lifetime; intentionally omitted. }, [form.status]); - const handleSubmit = (e: React.FormEvent) => { + const handleSubmit = (e: React.FormEvent) => { e.preventDefault(); const problem = emailError(email); setEmailMessage(problem); @@ -131,7 +132,7 @@ export function AnnouncementToast({ document.getElementById('toast-email')?.focus(); return; } - void form.submit({ email: email.trim(), paper: 'overview' }); + void form.submit({ website_url: readHoneypot(e.currentTarget), email: email.trim(), paper: 'overview' }); }; if (!visible) return null; @@ -200,6 +201,7 @@ export function AnnouncementToast({ data-compact="" noValidate > + (null); - const form = useGrowthForm<{ email: string }>({ + const form = useGrowthForm<{ website_url: string; email: string }>({ route: '/api/newsletter', formPolicy, events: { @@ -38,7 +39,7 @@ function NewsletterForm({ formPolicy }: { formPolicy: PublicFormPolicy }) { }); const disclosureId = 'footer-newsletter-growth-disclosure'; - const handleSubmit = (e: React.FormEvent) => { + const handleSubmit = (e: React.FormEvent) => { e.preventDefault(); const problem = emailError(email); setEmailMessage(problem); @@ -46,7 +47,7 @@ function NewsletterForm({ formPolicy }: { formPolicy: PublicFormPolicy }) { document.getElementById('footer-email')?.focus(); return; } - void form.submit({ email: email.trim() }); + void form.submit({ website_url: readHoneypot(e.currentTarget), email: email.trim() }); }; if (form.status === 'sent') { @@ -71,6 +72,7 @@ function NewsletterForm({ formPolicy }: { formPolicy: PublicFormPolicy }) { return ( +
({})); +import { createLeadRoute } from '../../app/api/leads/route'; +import { createNewsletterRoute } from '../../app/api/newsletter/route'; +import { createWhitepaperSignupRoute } from '../../app/api/whitepaper-signup/route'; +// eslint-disable-next-line @nx/enforce-module-boundaries +import { FormRateLimitError } from '@threadplane-internal/growth'; +import { getFormPolicy } from './form-policy'; +import type { GrowthFormRouteDependencies } from './form-route'; + +const policy = getFormPolicy({ GROWTH_FORM_POLICY: 'growth_v1' }); +const submissionId = '20000000-0000-4000-8000-000000000002'; + +describe.each([ + ['contact', createLeadRoute], + ['pricing', createLeadRoute], + ['newsletter', createNewsletterRoute], + ['whitepaper', createWhitepaperSignupRoute], +] as const)('%s abuse route', (kind, createRoute) => { + function harness() { + const accept = vi + .fn() + .mockResolvedValue({ + accepted: true, + approved: true, + contactId: 'contact', + submissionId, + }); + const deps: GrowthFormRouteDependencies = { + accept, + getPolicy: () => policy, + now: () => new Date('2026-09-09T14:00:00Z'), + loadKeyring: () => ({ + active: { + version: 1, + secret: 'route-test-key-that-is-at-least-32-bytes', + }, + }), + createDatabase: vi.fn().mockReturnValue({ close: vi.fn() }), + nudge: vi.fn().mockResolvedValue(undefined), + }; + const request = (extra: Record = {}) => + new Request('https://threadplane.ai/api/forms', { + method: 'POST', + headers: { + 'content-type': 'application/json', + 'x-forwarded-for': '203.0.113.1', + }, + body: JSON.stringify({ + submission_id: submissionId, + policy_version: policy.version, + email: 'reader@gmail.com', + form_kind: kind, + ...extra, + }), + }); + return { accept, deps, request, POST: createRoute(deps).POST }; + } + it('acknowledges blocked forms silently without dispatching', async () => { + const h = harness(); + h.accept.mockResolvedValue({ + accepted: true, + approved: false, + contactId: 'contact', + submissionId, + deliverySuppressed: true, + }); + const response = await h.POST( + h.request({ + website_url: 'https://spam.invalid', + score: 0, + decision: 'allow', + }) + ); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ ok: true }); + expect(h.deps.nudge).not.toHaveBeenCalled(); + const input = h.accept.mock.calls[0][1]; + expect(input.honeypot).toBe('https://spam.invalid'); + expect(input.score).toBeUndefined(); + expect(input.trustedClientIp).toBeUndefined(); + }); + it('returns Retry-After for a rate limit without dispatching', async () => { + const h = harness(); + h.accept.mockRejectedValue(new FormRateLimitError(120)); + const response = await h.POST(h.request()); + expect(response.status).toBe(429); + expect(response.headers.get('Retry-After')).toBe('120'); + expect(h.deps.nudge).not.toHaveBeenCalled(); + }); + it('keeps ordinary and older submissions without honeypot working', async () => { + const h = harness(); + expect((await h.POST(h.request())).status).toBe(200); + expect(h.deps.nudge).toHaveBeenCalledOnce(); + }); + it('continues a borderline submission admitted by the server', async () => { + const h = harness(); + expect((await h.POST(h.request({ email: 'test@example.com' }))).status).toBe(200); + expect(h.accept.mock.calls[0][1].email).toBe('test@example.com'); + expect(h.deps.nudge).toHaveBeenCalledOnce(); + }); + it('preserves a silent blocked decision on retry', async () => { + const h = harness(); + h.accept.mockResolvedValue({ + accepted: true, + approved: false, + contactId: 'contact', + submissionId, + deliverySuppressed: true, + }); + for (let attempt = 0; attempt < 2; attempt++) { + const response = await h.POST(h.request({ website_url: 'bot' })); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ ok: true }); + } + expect(h.deps.nudge).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/website/src/lib/growth/form-route.ts b/apps/website/src/lib/growth/form-route.ts index 402b968e6..88a901e31 100644 --- a/apps/website/src/lib/growth/form-route.ts +++ b/apps/website/src/lib/growth/form-route.ts @@ -1,9 +1,11 @@ import 'server-only'; +import { isIP } from 'node:net'; // The website intentionally consumes the growth library through its internal boundary. // eslint-disable-next-line @nx/enforce-module-boundaries import { acceptFormSubmission, + FormRateLimitError, createDatabaseExecutor, type AcceptFormSubmissionInput, type AcceptFormSubmissionResult, @@ -11,6 +13,23 @@ import { type SqlExecutor, } from '@threadplane-internal/growth'; +export function trustedFormClientIp(request: Request): string | undefined { + // Only trust the platform-overwritten header inside Vercel. Never use an + // arbitrary forwarded-for header supplied to a local/custom-hosted server. + if (process.env['VERCEL'] !== '1') return undefined; + const ip = request.headers.get('x-vercel-forwarded-for')?.trim(); + return ip && isIP(ip) ? ip.toLowerCase() : undefined; +} + +export function formAdmissionError(error: unknown): Response | undefined { + if (!(error instanceof FormRateLimitError)) return undefined; + return jsonResponse( + { error: 'Please try again later', retryable: true }, + 429, + { 'Retry-After': String(error.retryAfterSec) } + ); +} + import { readBoundedBody } from '../../app/api/_internal/read-bounded-body'; import { loadEmailHmacKeyring } from './email-keyring'; import { getFormPolicy, type PublicFormPolicy } from './form-policy'; diff --git a/apps/website/src/styles/forms.css b/apps/website/src/styles/forms.css index 5819fdf7b..82f823a5e 100644 --- a/apps/website/src/styles/forms.css +++ b/apps/website/src/styles/forms.css @@ -234,3 +234,11 @@ select[data-ui="form-control"] { transition: none; } } +.form-honeypot { + position: absolute; + width: 1px; + height: 1px; + overflow: hidden; + clip-path: inset(50%); + pointer-events: none; +} diff --git a/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-cutover.md b/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-cutover.md index e3e3d8848..3045ca3d4 100644 --- a/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-cutover.md +++ b/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-cutover.md @@ -277,13 +277,22 @@ Any legacy side effect, campaign job, scheduled provider follow-up, or missing d ### PRODUCTION LIVE — explicit authorization required: sender identity +For the recipient-only delivery rollout, deploy website webhook Message-ID +binding support first. Verify exact provider/job binding with existing accepted +delivery records before deploying lifecycle BCC removal and its bounded lookup +fallback. Keep campaign switches unchanged. An unresolved binding must defer +only that contact's follow-ups; it must never trigger resending an accepted email. + From one received allowlisted message, verify and record pass/fail without copying raw headers: - SPF alignment/pass, DKIM alignment/pass, and DMARC pass for `threadplane.ai`; - expected Return-Path; - `List-Unsubscribe` with the opaque HTTPS action URL; - `List-Unsubscribe-Post: List-Unsubscribe=One-Click`; -- Brian BCC seed and `X-Threadplane-Job-ID` on the received copy; +- no recipient BCC, and `X-Threadplane-Job-ID` on the received message; +- the provider's actual RFC Message-ID bound to the exact accepted growth job, + including a matched reply stopping follow-ups and a missing binding delaying + only that contact's follow-ups; - `Reply-To` routes replies to Brian; - no open pixel and no click-link rewriting. diff --git a/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-operations.md b/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-operations.md index 44592d830..c0e3dd0c6 100644 --- a/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-operations.md +++ b/docs/superpowers/runbooks/2026-08-31-growth-lifecycle-operations.md @@ -53,10 +53,60 @@ Repeat every protected health/stop/sender gate before following the same switch ## Runtime invariants +Provider binding uses Resend's documented sent-email `message_id` field +([July 2026 provider update](https://resend.com/changelog/message-id-for-sent-emails)). +The exact sent-email lookup also validates its environment tag before binding. + +Public growth forms record a server-owned `form.abuse_assessed` activity with +rule version, score, category, and reasons. `form-abuse-v1` blocks at 80/100; +borderline submissions proceed normally. A filled invisible honeypot scores 100. +Repeated mixed-case gibberish across name/message and corroborating company or +identity patterns contribute independently. Placeholder addresses alone do not +cross the conservative threshold. + +Blocked submissions retain evidence without creating delivery, founder +notification, enrichment, or campaign jobs. They do not overwrite an existing +legitimate contact's profile or consent. There is no blocked-submission digest. +Manual suppressions remain in force for later submissions from the same address. + +The shared database admits at most five new submissions per normalized email +and twenty per trusted platform IP in each hourly window. Counter keys use HMAC; +no raw IP is retained. Immutable submission retries consume no additional quota. +A denied request returns 429 with Retry-After; a counter-storage failure rolls +back its savepoint and continues local scoring, recording +`limiter_unavailable=true` for operator review. + Duplicate cron invocations are normal. Skip-locked leases, lease tokens, immutable activity keys, job idempotency keys, and Resend idempotency keys must yield at most one effect. An unmatched `mailbox.recovery_required` blocks `send_step` and `reply_reconcile` leasing and final submission. Recovery-safe non-mail work may continue. Work resumes only after the matching `mailbox.recovery_completed` event. +Recipient-facing growth emails have no BCC. Separate internal notifications still +target the configured founder, with the existing non-production allowlist check. +The verified sender, Reply-To, unsubscribe headers, recipient allowlist, and +provider idempotency requirements remain in effect. + +Reply matching uses the actual RFC Message-ID from a verified Resend webhook, +matched to the exact accepted provider email ID and database environment. Existing +Gmail seed bindings remain valid. Duplicate bindings revisit pending reply +reconciliation; conflicting identifiers cannot replace an existing binding. + +Before leasing, the lifecycle dispatcher reserves at most five missing bindings +for authenticated GET `/emails/:id` lookups. Each lookup has a five-second timeout +and a 128 KiB response limit. Accepted jobs retain `message_id_lookup_attempts` +and `message_id_lookup_after` in their payload; retries back off from two minutes +to an hourly maximum. Missing or invalid identities and lookup failures leave the +send completed and retryable for binding only. Never reopen or resubmit an +accepted email to recover its Message-ID. Missing delivery configuration skips +lookups so non-mail dispatch can continue; configured environment mismatches and +database failures remain errors. + +A contact with an accepted but unbound prior recipient send must wait before a +campaign follow-up; final authorization returns `reply_binding_pending` and the +worker defers the follow-up. Other contacts continue under the existing campaign +switches. Investigate persistent unbound jobs using their exact provider IDs, +environment configuration, retry state, and webhook delivery history. Do not +clear the binding gate or disable the whole campaign to bypass this condition. + The worker checks cancellation after asynchronous preparation and before recipient submission, internal at-most-once claims, and provider calls. Once a provider call begins, settle its known/rejected/ambiguous outcome even if the request later aborts. Never automatically resubmit an expired lease with final authorization or a prior internal submission claim. Deterministically corrupt persisted input becomes `deterministic_job_poison` and does not stop the remaining leased batch. Abort, heartbeat loss, and infrastructure errors stop the batch and must not be misclassified as poison. diff --git a/libs/growth/src/index.ts b/libs/growth/src/index.ts index c41125e4e..8a35ac056 100644 --- a/libs/growth/src/index.ts +++ b/libs/growth/src/index.ts @@ -51,3 +51,7 @@ export { readGrowthFunnel, readContactJourney, } from './lib/observability/journey-report.ts'; +export { FormRateLimitError } from './lib/form-admission.ts'; +export { assessFormAbuse, FORM_ABUSE_THRESHOLD, FORM_ABUSE_VERSION, type FormAbuseAssessment } from './lib/form-abuse.ts'; +export { bindProviderMessageId } from './lib/replies.ts'; +export { reconcilePendingResendMessageIds } from './lib/resend.ts'; diff --git a/libs/growth/src/lib/contacts.spec.ts b/libs/growth/src/lib/contacts.spec.ts index 0f6fde79f..7bb114ef7 100644 --- a/libs/growth/src/lib/contacts.spec.ts +++ b/libs/growth/src/lib/contacts.spec.ts @@ -348,6 +348,23 @@ describe('install/runtime contact approval', () => { }); describe('approveContactFromForm', () => { + it('records blocked facts without changing a legitimate existing contact or consent', async () => { + const approved = contactRow({ outreach_approved_at: occurredAt }); + const harness = executorWith({ + 'lock-email': () => ({ rows: [] }), 'find-contact': () => ({ rows: [approved] }), + 'find-hard-stops': () => ({ rows: [] }), + 'insert-activity': (parameters) => { + expect(JSON.parse(String(parameters[4]))).toMatchObject({ approval_granted: false, form_abuse_blocked: true }); + return { rows: [{ event_key: baseApproval.eventKey }] }; + }, + 'read-control-state': () => ({ rows: [{ ...approved, latest_hard_stop_kind: null, latest_hard_stop_at: null }] }), + }); + const result = await approveContactFromForm(harness.executor, { ...baseApproval, serverFormBlocked: true }); + expect(result.formApprovalGranted).toBe(false); + expect(result.canSend).toBe(true); + expect(harness.calls.some(c => ['update-contact-facts', 'set-form-approval', 'insert-form-outreach-approved'].includes(c.marker))).toBe(false); + }); + it('normalizes direct facts, preserves a private lookup, and records exact approval provenance', async () => { const harness = executorWith({ 'lock-email': () => ({ rows: [{}] }), diff --git a/libs/growth/src/lib/contacts.ts b/libs/growth/src/lib/contacts.ts index 5cd0c9118..f579f4a45 100644 --- a/libs/growth/src/lib/contacts.ts +++ b/libs/growth/src/lib/contacts.ts @@ -119,6 +119,8 @@ export interface FormApprovalControlState extends ContactControlState { } export interface ApproveContactFromFormInput { + /** Server-owned form admission; never accept this field from a request body. */ + serverFormBlocked?: boolean; email: string; displayName?: string | null; companyName?: string | null; @@ -639,6 +641,7 @@ function canonicalJson(value: unknown): string { } interface PreparedFormApproval { + formBlocked: boolean; activeLookup: ReturnType[number]; candidates: ReturnType; companyDomain: string | null; @@ -708,6 +711,7 @@ function prepareFormApproval( ); const submittedFacts = input.submittedFacts ?? {}; const formRequestData = { + ...(input.serverFormBlocked ? { form_abuse_blocked: true } : {}), company_domain: companyDomain, company_name: companyName, display_name: displayName, @@ -722,6 +726,7 @@ function prepareFormApproval( }; return { + formBlocked: input.serverFormBlocked === true, activeLookup, candidates, companyDomain, @@ -743,6 +748,7 @@ async function approvePreparedContactFromForm( prepared: PreparedFormApproval ): Promise { const { + formBlocked, activeLookup, candidates, companyDomain, @@ -939,7 +945,7 @@ async function approvePreparedContactFromForm( latestHardStopAt !== null && (approvedAt === null || latestHardStopAt.getTime() >= approvedAt.getTime()); const currentlyAuthorized = currentlyApproved && !stoppedAfterApproval; - const approvalAllowed = currentlyAuthorized || latestHardStop == null; + const approvalAllowed = !formBlocked && (currentlyAuthorized || latestHardStop == null); const activityInserted = await insertActivityOnce(transaction, { eventKey, contactId: contact.id, @@ -1009,7 +1015,7 @@ async function approvePreparedContactFromForm( ); } - if (!currentlyApproved && latestHardStop == null) { + if (approvalAllowed && !currentlyApproved && latestHardStop == null) { await transaction.execute( `/* growth:set-form-approval */ update growth_contacts diff --git a/libs/growth/src/lib/form-abuse.spec.ts b/libs/growth/src/lib/form-abuse.spec.ts new file mode 100644 index 000000000..7468b8096 --- /dev/null +++ b/libs/growth/src/lib/form-abuse.spec.ts @@ -0,0 +1,83 @@ +import { describe, expect, it } from 'vitest'; +import { assessFormAbuse } from './form-abuse.ts'; + +describe('form abuse assessment', () => { + it('blocks combined random name and message signals', () => { + expect( + assessFormAbuse({ + email: 'person@gmail.com', + displayName: 'aBcDeFgHiJkLmNoP', + message: 'qRsTuVwXyZaBcDeF', + companyName: 'Zqxwy LLC', + }) + ).toMatchObject({ + score: 100, + decision: 'blocked', + category: 'automated_form_spam', + reasons: ['gibberish_name', 'gibberish_message', 'generated_company'], + }); + }); + it.each([ + { + displayName: 'Christopher', + message: 'Can you help us integrate Angular agents?', + companyName: 'Acme LLC', + }, + { displayName: '李明', message: '想了解你们的产品', companyName: '示例' }, + { displayName: 'McDonald', message: 'Hello', companyName: 'NASA' }, + { + displayName: 'Jean-Baptiste Martin', + message: 'Please send more information.', + companyName: 'Test LLC', + }, + { + displayName: 'JOHNATHANSMITH', + message: 'HelloWorldExample', + companyName: 'ACME LLC', + }, + ])( + 'allows legitimate short and international fields: $displayName', + (fields) => { + expect( + assessFormAbuse({ email: 'hello+demo@gmail.com', ...fields }) + ).toMatchObject({ score: 0, decision: 'allow', category: 'normal' }); + } + ); + it('keeps a single gibberish signal plus LLC below threshold', () => { + expect( + assessFormAbuse({ + email: 'person@gmail.com', + displayName: 'aBcDeFgHiJkLmNoP', + companyName: 'Zqxwy LLC', + }) + ).toMatchObject({ score: 60, decision: 'allow', category: 'suspicious' }); + }); + it('blocks placeholder identities only with corroborating evidence', () => { + expect(assessFormAbuse({ email: 'test@example.com' })).toMatchObject({ + score: 60, + decision: 'allow', + }); + expect( + assessFormAbuse({ email: 'test@example.com', rapidIdentityChange: true }) + ).toMatchObject({ + score: 80, + decision: 'blocked', + category: 'placeholder_email', + }); + expect(assessFormAbuse({ email: 'test@realbusiness.com' })).toMatchObject({ + score: 0, + decision: 'allow', + }); + }); + it('blocks the honeypot alone and accepts older forms without it', () => { + expect( + assessFormAbuse({ + email: 'person@gmail.com', + honeypot: 'https://bot.invalid', + }) + ).toMatchObject({ score: 100, decision: 'blocked', category: 'honeypot' }); + expect( + assessFormAbuse({ email: 'person@gmail.com', honeypot: ' ' }).decision + ).toBe('allow'); + }); +}); diff --git a/libs/growth/src/lib/form-abuse.ts b/libs/growth/src/lib/form-abuse.ts new file mode 100644 index 000000000..ffacbc168 --- /dev/null +++ b/libs/growth/src/lib/form-abuse.ts @@ -0,0 +1,90 @@ +export const FORM_ABUSE_VERSION = 'form-abuse-v1'; +export const FORM_ABUSE_THRESHOLD = 80; + +export interface FormAbuseAssessment { + version: typeof FORM_ABUSE_VERSION; + score: number; + decision: 'allow' | 'blocked'; + category: + | 'normal' + | 'suspicious' + | 'automated_form_spam' + | 'placeholder_email' + | 'honeypot'; + reasons: string[]; +} + +export interface FormAbuseFacts { + email: string; + displayName?: string | null; + companyName?: string | null; + message?: string | null; + honeypot?: string; + rapidIdentityChange?: boolean; +} + +// Deliberately narrow: ordinary words, acronyms, PascalCase and international +// names do not have repeated upper/lower case changes inside a long token. +function randomToken(value: string | null | undefined): boolean { + const text = value?.trim() ?? ''; + if (!/^[A-Za-z]{12,64}$/.test(text)) return false; + const upper = (text.match(/[A-Z]/g) ?? []).length; + if (upper < 3 || text.length - upper < 3) return false; + let transitions = 0; + for (let index = 1; index < text.length; index++) { + if (/[A-Z]/.test(text[index]) !== /[A-Z]/.test(text[index - 1])) + transitions++; + } + return transitions >= 6; +} + +export function assessFormAbuse(facts: FormAbuseFacts): FormAbuseAssessment { + const reasons: string[] = []; + let score = 0; + if (randomToken(facts.displayName)) { + reasons.push('gibberish_name'); + score += 40; + } + if (randomToken(facts.message)) { + reasons.push('gibberish_message'); + score += 40; + } + if (score && /^[A-Za-z]{4,12} LLC$/.test(facts.companyName?.trim() ?? '')) { + reasons.push('generated_company'); + score += 20; + } + const placeholder = + /^(?:test|testing|dummy|fake|asdf|nobody)(?:[0-9]*)@(?:example\.(?:com|org|net)|test\.com|invalid\.com)$/i.test( + facts.email.trim() + ); + if (placeholder) { + reasons.push('placeholder_email'); + score += 60; + } + if (facts.rapidIdentityChange) { + reasons.push('rapid_identity_change'); + score += 20; + } + const honeypot = Boolean(facts.honeypot?.trim()); + if (honeypot) { + reasons.push('honeypot'); + score = 100; + } + score = Math.min(score, 100); + const blocked = score >= FORM_ABUSE_THRESHOLD; + return { + version: FORM_ABUSE_VERSION, + score, + decision: blocked ? 'blocked' : 'allow', + category: honeypot + ? 'honeypot' + : blocked + ? placeholder + ? 'placeholder_email' + : 'automated_form_spam' + : score + ? 'suspicious' + : 'normal', + reasons, + }; +} diff --git a/libs/growth/src/lib/form-admission.spec.ts b/libs/growth/src/lib/form-admission.spec.ts new file mode 100644 index 000000000..4d67ba746 --- /dev/null +++ b/libs/growth/src/lib/form-admission.spec.ts @@ -0,0 +1,136 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + prepareFormAdmission, + recordFormAdmission, + FormRateLimitError, +} from './form-admission.ts'; +import type { SqlTransaction } from './database.ts'; + +const input = { + submissionId: '20000000-0000-4000-8000-000000000002', + email: 'person@gmail.com', + form: { kind: 'contact' as const, message: 'Hello there' }, + source: 'website', + sourceForm: 'contact', + noticeText: 'Follow up', + noticeVersion: 'v1', + policyVersion: 'v1', + keyring: { + active: { version: 1, secret: 'a-long-admission-secret-for-tests-only' }, + }, + occurredAt: new Date('2026-09-09T14:01:00Z'), +}; + +function harness(count = 1, unavailable = false) { + const queries: string[] = []; + const tx: SqlTransaction = { + execute: vi.fn(async (q: string) => { + queries.push(q); + if (q.includes('growth:consume-form-budget')) { + if (unavailable) throw new Error('provider secret should not leak'); + return { rows: [{ count }] }; + } + return { rows: [] }; + }) as SqlTransaction['execute'], + }; + return { tx, queries }; +} + +describe('transactional form admission', () => { + it('replays the immutable decision without consuming a budget and rejects changed identities', async () => { + const h = harness(); + const first = await prepareFormAdmission(h.tx, input); + const result = { + accepted: true as const, + approved: true, + contactId: 'contact', + submissionId: input.submissionId, + }; + await recordFormAdmission(h.tx, input, first, result); + const call = vi + .mocked(h.tx.execute) + .mock.calls.find(([q]) => q.includes('record-form-assessment')); + const data = JSON.parse(String(call?.[1]?.[3])); + const tx: SqlTransaction = { + execute: vi.fn(async (q: string) => ({ + rows: q.includes('read-form-assessment') ? [{ data }] : [], + })) as SqlTransaction['execute'], + }; + expect((await prepareFormAdmission(tx, input)).replay).toEqual(result); + expect( + vi + .mocked(tx.execute) + .mock.calls.some(([q]) => q.includes('consume-form-budget')) + ).toBe(false); + await expect( + prepareFormAdmission(tx, { ...input, email: 'other@gmail.com' }) + ).rejects.toThrow('identity conflict'); + }); + + it('corroborates a placeholder with a recent different identity', async () => { + const h = harness(); + const original = h.tx.execute; + h.tx.execute = vi.fn(async (q: string, p: readonly unknown[] = []) => + q.includes('read-form-history') + ? { + rows: [ + { + display_name: 'Earlier Person', + company_name: 'Earlier Company', + }, + ], + } + : original(q, p) + ) as SqlTransaction['execute']; + expect( + ( + await prepareFormAdmission(h.tx, { + ...input, + email: 'test@example.com', + displayName: 'New Person', + companyName: 'New Company', + }) + ).assessment + ).toMatchObject({ + score: 80, + decision: 'blocked', + category: 'placeholder_email', + }); + }); + + it('limits the sixth new email submission with a bounded retry time', async () => { + const h = harness(6); + await expect(prepareFormAdmission(h.tx, input)).rejects.toMatchObject({ + retryAfterSec: 3540, + }); + expect( + h.queries.some((q) => q.includes('savepoint growth_form_budget')) + ).toBe(true); + }); + it('continues local scoring after rolling back a failed budget operation', async () => { + const h = harness(1, true); + const result = await prepareFormAdmission(h.tx, input); + expect(result).toMatchObject({ + limiterUnavailable: true, + assessment: { decision: 'allow' }, + }); + expect(h.queries).toContain('rollback to savepoint growth_form_budget'); + }); + it('does not mistake a denied budget for an infrastructure failure', async () => { + const h = harness(21); + await expect( + prepareFormAdmission(h.tx, { ...input, trustedClientIp: '203.0.113.5' }) + ).rejects.toBeInstanceOf(FormRateLimitError); + }); + it('locks a normalized identity before inspecting history', async () => { + const h = harness(); + await prepareFormAdmission(h.tx, { ...input, email: ' Person@Gmail.com ' }); + expect(h.tx.execute).toHaveBeenCalledWith( + expect.stringContaining('growth:lock-form-email'), + ['person@gmail.com'] + ); + expect( + h.queries.findIndex((q) => q.includes('lock-form-email')) + ).toBeLessThan(h.queries.findIndex((q) => q.includes('read-form-history'))); + }); +}); diff --git a/libs/growth/src/lib/form-admission.ts b/libs/growth/src/lib/form-admission.ts new file mode 100644 index 000000000..619592d8a --- /dev/null +++ b/libs/growth/src/lib/form-admission.ts @@ -0,0 +1,213 @@ +import { createHmac } from 'node:crypto'; +import { normalizeRecipientEmail } from './crypto.ts'; +import type { SqlTransaction } from './database.ts'; +import type { + AcceptFormSubmissionInput, + AcceptFormSubmissionResult, +} from './forms.ts'; +import { assessFormAbuse, type FormAbuseAssessment } from './form-abuse.ts'; + +export class FormRateLimitError extends Error { + constructor(readonly retryAfterSec: number) { + super('Form submission rate limited'); + this.name = 'FormRateLimitError'; + } +} + +export interface FormAdmission { + assessment: FormAbuseAssessment; + requestDigest: string; + limiterUnavailable: boolean; + replay?: AcceptFormSubmissionResult; +} + +function canonical(value: unknown): string { + function sorted(value: unknown): unknown { + if (Array.isArray(value)) return value.map(sorted); + if (value && typeof value === 'object') + return Object.fromEntries( + Object.entries(value) + .filter(([, v]) => v !== undefined) + .sort(([a], [b]) => a.localeCompare(b)) + .map(([k, v]) => [k, sorted(v)]) + ); + return value; + } + return JSON.stringify(sorted(value)); +} + +export async function prepareFormAdmission( + tx: SqlTransaction, + input: AcceptFormSubmissionInput +): Promise { + const email = normalizeRecipientEmail(input.email); + const digest = (secret: string | Uint8Array, value: string) => + createHmac('sha256', secret).update(value).digest('hex'); + const request = canonical({ + email, + name: input.displayName?.trim() || null, + company: input.companyName?.trim() || null, + form: input.form, + source: input.source, + sourceForm: input.sourceForm, + noticeText: input.noticeText, + noticeVersion: input.noticeVersion, + policyVersion: input.policyVersion, + session: input.acquisitionSessionId || null, + honeypot: input.honeypot?.trim() || null, + }); + const requestDigest = digest( + input.keyring.active.secret, + `form-request:${request}` + ); + const eventKey = `form:${input.submissionId}:assessment`; + await tx.execute( + '/* growth:lock-form-submission */ select pg_advisory_xact_lock(hashtextextended($1, 0))', + [eventKey] + ); + await tx.execute( + '/* growth:lock-form-email */ select pg_advisory_xact_lock(hashtextextended($1, 0))', + [email] + ); + const prior = await tx.execute<{ data: Record }>( + '/* growth:read-form-assessment */ select data from growth_activity where event_key = $1', + [eventKey] + ); + if (prior.rows[0]) { + const data = prior.rows[0].data; + const key = [input.keyring.active, ...(input.keyring.previous ?? [])].find( + (k) => k.version === data['key_version'] + ); + if ( + !key || + digest(key.secret, `form-request:${request}`) !== data['request_digest'] + ) { + throw new Error('Form submission identity conflict'); + } + return { + assessment: data['assessment'] as FormAbuseAssessment, + replay: data['result'] as AcceptFormSubmissionResult, + requestDigest, + limiterUnavailable: data['limiter_unavailable'] === true, + }; + } + const history = await tx.execute<{ + display_name: string | null; + company_name: string | null; + }>( + `/* growth:read-form-history */ + select a.data->>'display_name' display_name, a.data->>'company_name' company_name + from growth_activity a join growth_contacts c on c.id = a.contact_id + where c.email_normalized = $1 and a.kind = 'contact.form_submission' + and a.occurred_at >= $2::timestamptz - interval '1 minute' + and a.occurred_at <= $2 and a.data->>'submission_id' <> $3 + order by a.occurred_at desc limit 10`, + [email, input.occurredAt, input.submissionId] + ); + const name = input.displayName?.trim(); + const company = input.companyName?.trim(); + const rapidIdentityChange = Boolean( + name && + company && + history.rows.some( + (row) => + row.display_name && + row.company_name && + row.display_name !== name && + row.company_name !== company + ) + ); + const assessment = assessFormAbuse({ + email, + displayName: name, + companyName: company, + message: 'message' in input.form ? input.form.message : undefined, + honeypot: input.honeypot, + rapidIdentityChange, + }); + + // Accepted legacy retries predate assessment records. Their existing immutable + // ledger validates facts later, but they must not spend another budget slot. + const legacy = await tx.execute<{ event_key: string }>( + "/* growth:read-legacy-form-retry */ select event_key from growth_activity where event_key = $1 and kind = 'contact.form_submission'", + [`form:${input.submissionId}:accepted`] + ); + let limiterUnavailable = false; + if (!legacy.rows.length) { + const now = input.occurredAt.getTime(); + const window = new Date(Math.floor(now / 3_600_000) * 3_600_000); + const buckets = [ + { + key: `form:email:${digest( + input.keyring.active.secret, + `form-email:${email}` + )}`, + limit: 5, + }, + ]; + if (input.trustedClientIp) + buckets.push({ + key: `form:ip:${digest( + input.keyring.active.secret, + `form-ip:${input.trustedClientIp}` + )}`, + limit: 20, + }); + await tx.execute('savepoint growth_form_budget'); + try { + for (const bucket of buckets.sort((a, b) => a.key.localeCompare(b.key))) { + const count = await tx.execute<{ count: string | number }>( + `/* growth:consume-form-budget */ + insert into growth_collection_budgets(bucket_key, window_start, count) values($1,$2,1) + on conflict(bucket_key,window_start) do update set count=growth_collection_budgets.count+1 returning count`, + [bucket.key, window] + ); + if (!count.rows[0]) throw new Error('Form budget unavailable'); + if (Number(count.rows[0].count) > bucket.limit) + throw new FormRateLimitError( + Math.max(1, Math.ceil((window.getTime() + 3_600_000 - now) / 1000)) + ); + } + await tx.execute( + `/* growth:expire-form-budgets */ + delete from growth_collection_budgets where (bucket_key,window_start) in + (select bucket_key,window_start from growth_collection_budgets where bucket_key like 'form:%' + and window_start < $1::timestamptz - interval '1 day' limit 100)`, + [window] + ); + await tx.execute('release savepoint growth_form_budget'); + } catch (error) { + await tx.execute('rollback to savepoint growth_form_budget'); + await tx.execute('release savepoint growth_form_budget'); + if (error instanceof FormRateLimitError) throw error; + limiterUnavailable = true; + } + } + return { assessment, requestDigest, limiterUnavailable }; +} + +export async function recordFormAdmission( + tx: SqlTransaction, + input: AcceptFormSubmissionInput, + admission: FormAdmission, + result: AcceptFormSubmissionResult +): Promise { + await tx.execute( + `/* growth:record-form-assessment */ + insert into growth_activity(event_key,contact_id,kind,occurred_at,data) + values($1,$2,'form.abuse_assessed',$3,$4::jsonb)`, + [ + `form:${input.submissionId}:assessment`, + result.contactId, + input.occurredAt, + JSON.stringify({ + assessment: admission.assessment, + result, + submission_id: input.submissionId, + request_digest: admission.requestDigest, + key_version: input.keyring.active.version, + limiter_unavailable: admission.limiterUnavailable, + }), + ] + ); +} diff --git a/libs/growth/src/lib/forms.spec.ts b/libs/growth/src/lib/forms.spec.ts index 8cff4464d..5d380bcfa 100644 --- a/libs/growth/src/lib/forms.spec.ts +++ b/libs/growth/src/lib/forms.spec.ts @@ -93,6 +93,47 @@ const baseInput = { }; describe('acceptFormSubmission', () => { + it('records high-confidence junk but creates no work or approval', async () => { + const harness = createHarness('stopped'); + const result = await acceptFormSubmission( + harness.executor, + { + ...baseInput, + displayName: 'aBcDeFgHiJkLmNoP', + companyName: 'Zqxwy LLC', + form: { kind: 'contact', message: 'qRsTuVwXyZaBcDeF' }, + }, + { approveContact: harness.approveContact } + ); + expect(harness.approveContact).toHaveBeenCalledWith( + harness.transaction, + expect.objectContaining({ serverFormBlocked: true }) + ); + expect(result).toMatchObject({ approved: false, deliverySuppressed: true }); + expect(harness.insertedJobKeys).toEqual([]); + expect( + harness.queries.some((q) => + q.sql.includes('growth:record-form-assessment') + ) + ).toBe(true); + }); + + it('does not enqueue fulfillment for an operator-suppressed contact', async () => { + const harness = createHarness('stopped'); + harness.approveContact.mockResolvedValue({ + ...(await harness.approveContact()), + latestHardStop: { + reason: 'manual_suppression' as 'unsubscribe', + occurredAt, + }, + }); + const result = await acceptFormSubmission(harness.executor, baseInput, { + approveContact: harness.approveContact, + }); + expect(result).toMatchObject({ approved: false, deliverySuppressed: true }); + expect(harness.insertedJobKeys).toEqual([]); + }); + it('approves and enqueues fulfillment, enrichment, and notification in one transaction', async () => { const harness = createHarness(); diff --git a/libs/growth/src/lib/forms.ts b/libs/growth/src/lib/forms.ts index 7f1a93f96..a441e7e13 100644 --- a/libs/growth/src/lib/forms.ts +++ b/libs/growth/src/lib/forms.ts @@ -6,6 +6,7 @@ import { approveContactFromFormInTransaction } from './contacts.ts'; import type { EmailHmacKeyring } from './crypto.ts'; import type { SqlExecutor, SqlTransaction } from './database.ts'; import type { GrowthEmailClassification } from './models.ts'; +import { prepareFormAdmission, recordFormAdmission } from './form-admission.ts'; const UUID_V4 = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu; @@ -45,6 +46,8 @@ export interface AcceptFormSubmissionInput { occurredAt: Date; keyring: EmailHmacKeyring; serverEmailClassification?: GrowthEmailClassification; + honeypot?: string; + trustedClientIp?: string; } export interface AcceptFormSubmissionResult { @@ -52,6 +55,7 @@ export interface AcceptFormSubmissionResult { approved: boolean; contactId: string; submissionId: string; + deliverySuppressed?: boolean; } interface AcceptFormSubmissionDependencies { @@ -133,8 +137,12 @@ export async function acceptFormSubmission( ): Promise { const submissionId = uuid('submissionId', input.submissionId); const facts = submittedFacts(input, submissionId); + input = { ...input, submissionId }; return executor.transaction(async (transaction) => { + const admission = await prepareFormAdmission(transaction, input); + if (admission.replay) return admission.replay; + const blocked = admission.assessment.decision === 'blocked'; const contact = await dependencies.approveContact(transaction, { email: input.email, displayName: input.displayName, @@ -149,8 +157,27 @@ export async function acceptFormSubmission( keyring: input.keyring, serverEmailClassification: input.serverEmailClassification, submittedFacts: facts, + ...(blocked ? { serverFormBlocked: true } : {}), }); - const approved = contact.formApprovalGranted; + const approved = !blocked && contact.formApprovalGranted; + if ( + blocked || + (contact.authorization === 'stopped' && + contact.latestHardStop && + !['unsubscribe', 'campaign.reply_received'].includes( + contact.latestHardStop.reason + )) + ) { + const result: AcceptFormSubmissionResult = { + accepted: true, + approved: false, + contactId: contact.contactId, + submissionId, + deliverySuppressed: true, + }; + await recordFormAdmission(transaction, input, admission, result); + return result; + } const fulfillmentPayload = { form_kind: input.form.kind, ...(input.form.kind === 'whitepaper' ? { paper: input.form.paper } : {}), @@ -211,11 +238,13 @@ export async function acceptFormSubmission( throw new Error(`Growth form job idempotency conflict: ${submissionId}`); } - return { + const result: AcceptFormSubmissionResult = { accepted: true, approved, contactId: contact.contactId, submissionId, }; + await recordFormAdmission(transaction, input, admission, result); + return result; }); } diff --git a/libs/growth/src/lib/jobs.spec.ts b/libs/growth/src/lib/jobs.spec.ts index dd3d5599d..39bf8ba73 100644 --- a/libs/growth/src/lib/jobs.spec.ts +++ b/libs/growth/src/lib/jobs.spec.ts @@ -499,6 +499,23 @@ describe('job leasing', () => { }); describe('final fulfillment authorization', () => { + it.each([ + ['reply_binding_pending', { reply_binding_pending: true }], + ['contact_stopped', { form_abuse_blocked: true }], + ])('denies submission when %s', async (reason, flags) => { + const harness = executorWith({ + 'acquire-google-reconcile-advisory-lock': () => ({ rows: [] }), + 'lock-contact-for-send': () => ({ rows: [{ id: jobRow().contact_id, + email_normalized: 'reader@acme.com', outreach_approved_at: now, deleted_at: null, + latest_hard_stop_kind: null, latest_hard_stop_at: null, + campaign_approval_valid: true, campaign_enrollment_valid: true, ...flags }] }), + 'lock-job-for-send': () => ({ rows: [jobRow()] }), + }); + await expect(authorizeLeasedJobForSubmission(harness.executor, { + campaignEnabled: true, deliveryEnabled: true, jobId: String(jobRow().id), leaseToken, now, + })).resolves.toMatchObject({ authorized: false, reason }); + }); + it('requires the exact allowlisted approval event and immutable enrollment provenance for campaign sends', async () => { const harness = executorWith({ 'acquire-google-reconcile-advisory-lock': () => ({ rows: [{}] }), diff --git a/libs/growth/src/lib/jobs.ts b/libs/growth/src/lib/jobs.ts index 8a62423c9..29152cc09 100644 --- a/libs/growth/src/lib/jobs.ts +++ b/libs/growth/src/lib/jobs.ts @@ -102,7 +102,8 @@ export type FinalSendAuthorization = | 'campaign_disabled' | 'delivery_disabled' | 'outside_send_window' - | 'mailbox_recovery_required'; + | 'mailbox_recovery_required' + | 'reply_binding_pending'; job: GrowthJob; }; @@ -118,6 +119,8 @@ interface SendContactRow extends Record { fulfillment_deletion_blocked?: boolean; campaign_approval_valid?: boolean; campaign_enrollment_valid?: boolean; + reply_binding_pending?: boolean; + form_abuse_blocked?: boolean; } interface FinalSendAuthorizationRow extends Record { @@ -246,6 +249,16 @@ function installRuntimeApproval(alias: 'approval' | 'authoritative'): string { ))`; } +function blockedFormJob(alias: string): string { + return `exists (select 1 from growth_activity form_assessment + where form_assessment.kind = 'form.abuse_assessed' + and form_assessment.contact_id = ${alias}.contact_id + and form_assessment.event_key in ( + 'form:' || (${alias}.payload->>'submission_id') || ':assessment', + regexp_replace(${alias}.payload->>'approval_event_key', ':accepted:outreach-approved$', ':assessment') + ) and form_assessment.data->'result'->>'deliverySuppressed' = 'true')`; +} + export async function materializeCampaignEnrollment( executor: SqlExecutor, input: MaterializeCampaignEnrollmentInput @@ -317,6 +330,13 @@ export async function materializeCampaignEnrollment( from growth_activity approval where approval.contact_id = c.id and approval.occurred_at = c.outreach_approved_at + and not exists ( + select 1 from growth_activity form_assessment + where form_assessment.kind = 'form.abuse_assessed' + and form_assessment.contact_id = c.id + and form_assessment.event_key = regexp_replace(approval.event_key, ':accepted:outreach-approved$', ':assessment') + and form_assessment.data->'result'->>'deliverySuppressed' = 'true' + ) and ( ( approval.kind = 'form.outreach_approved' @@ -516,6 +536,13 @@ export async function leaseDueJobs( select j.id from growth_jobs j where j.kind = any($1::text[]) + and not ${blockedFormJob('j')} + and (j.kind <> 'send_step' or not exists ( + select 1 from growth_jobs unbound + where unbound.contact_id = j.contact_id and unbound.id <> j.id + and unbound.kind in ('fulfill', 'send_step') and unbound.status = 'completed' + and unbound.provider_email_id is not null and unbound.rfc_message_id is null + )) and ($5::boolean or j.kind <> 'send_step') and ( j.kind not in ('send_step', 'reply_reconcile') @@ -787,6 +814,13 @@ export async function authorizeLeasedJobForSubmission( c.email_normalized, c.outreach_approved_at, c.deleted_at, + ${blockedFormJob('target')} as form_abuse_blocked, + exists ( + select 1 from growth_jobs unbound + where unbound.contact_id = c.id and unbound.id <> target.id + and unbound.kind in ('fulfill', 'send_step') and unbound.status = 'completed' + and unbound.provider_email_id is not null and unbound.rfc_message_id is null + ) as reply_binding_pending, stop.kind as latest_hard_stop_kind, stop.occurred_at as latest_hard_stop_at, exists ( @@ -914,6 +948,12 @@ export async function authorizeLeasedJobForSubmission( if (contact.deleted_at !== null) { return { authorized: false, reason: 'contact_deleted', job }; } + if (contact.form_abuse_blocked === true) { + return { authorized: false, reason: 'contact_stopped', job }; + } + if (job.kind === 'send_step' && contact.reply_binding_pending === true) { + return { authorized: false, reason: 'reply_binding_pending', job }; + } if ( job.kind === 'fulfill' && (contact.fulfillment_delivery_blocked === true || @@ -1149,6 +1189,7 @@ export async function claimInternalNotificationSubmission( and j.status = 'leased' and j.lease_token = $2::uuid and j.lease_until > $3 + and not ${blockedFormJob('j')} on conflict (event_key) do nothing returning event_key`, [jobId, leaseToken, now] diff --git a/libs/growth/src/lib/replies.spec.ts b/libs/growth/src/lib/replies.spec.ts index 3398c75cb..fb73dd1de 100644 --- a/libs/growth/src/lib/replies.spec.ts +++ b/libs/growth/src/lib/replies.spec.ts @@ -9,6 +9,7 @@ import type { } from './database.ts'; import { GoogleReplyReplayError, + bindProviderMessageId, parseGoogleMailboxEvent, processGoogleMailboxEvent, isGoogleMailboxRecoveryPaused, @@ -2018,3 +2019,181 @@ describe('processGoogleMailboxEvent', () => { expect(harness.calls).toContain('complete-google-reconciled-reply'); }); }); + +describe('provider Message-ID binding', () => { + function bindingHarness( + overrides: TestRow = {}, + pending: TestRow[] = [], + conflictingJobs: TestRow[] = [] + ) { + const job = { + id: jobId, + kind: 'send_step', + contact_id: contactId, + status: 'completed', + provider_email_id: 'resend-email-1', + delivery_status: 'submitted', + rfc_message_id: null, + gmail_seed_message_id: null, + ...overrides, + }; + const bind = vi + .fn<(parameters: readonly unknown[]) => SqlQueryResult>() + .mockReturnValue({ rows: [{ id: jobId }] }); + const complete = vi.fn(() => ({ rows: [{ id: 'reconcile-1' }] })); + const revive = vi.fn(() => ({ rows: [{ id: 'reconcile-1' }] })); + const db = executorWith({ + ...commonHandlers(), + 'discover-provider-message-job': () => ({ rows: [job] }), + 'lock-provider-message-contact': () => ({ + rows: [{ id: contactId, deleted_at: null }], + }), + 'lock-provider-message-job': () => ({ rows: [job] }), + 'check-provider-message-conflicts': () => ({ rows: conflictingJobs }), + 'bind-provider-message-id': bind, + 'lock-google-reconcile-for-seed': (_parameters, sql) => ({ + rows: pending.filter( + (row) => + row['status'] !== 'failed' || + (sql.includes("last_error_code = 'founder_review'") && + row['last_error_code'] === 'founder_review') + ), + }), + 'revive-exhausted-google-reconcile': revive, + 'record-google-reconcile-candidate': () => ({ + rows: [{ id: 'reconcile-1' }], + }), + 'complete-google-reconciled-reply': complete, + }); + const stopContact = vi + .fn() + .mockResolvedValue({ applied: true, effective: true }); + return { ...db, bind, complete, revive, stopContact }; + } + const input = { + providerEmailId: 'resend-email-1', + rfcMessageId: '', + }; + function exhaustedReply(rank = 0, errorCode = 'founder_review'): TestRow { + return { + id: 'reconcile-1', + status: 'failed', + last_error_code: errorCode, + contact_id: null, + payload: { + gmail_message_id: 'reply-gmail', + occurred_at: now.toISOString(), + in_reply_to: + rank === 0 ? input.rfcMessageId : '', + references: [input.rfcMessageId], + ranked_candidates: [{ message_id: input.rfcMessageId, rank }], + resolved_candidates: [], + }, + }; + } + it('stops a direct reply that exhausted retries before its authenticated binding arrived', async () => { + const h = bindingHarness({}, [exhaustedReply()]); + await bindProviderMessageId(h.executor, input, { + stopContact: h.stopContact, + }); + expect(h.revive).toHaveBeenCalledOnce(); + expect(h.stopContact).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ contactId, reason: 'campaign.reply_received' }) + ); + expect(h.complete).toHaveBeenCalledOnce(); + }); + it('keeps an exhausted lower-ranked reply unresolved instead of opening its contact send gate', async () => { + const h = bindingHarness({}, [exhaustedReply(1)]); + await expect( + bindProviderMessageId(h.executor, input, { stopContact: h.stopContact }) + ).rejects.toThrow(/reconciliation conflict/u); + expect(h.revive).not.toHaveBeenCalled(); + expect(h.stopContact).not.toHaveBeenCalled(); + }); + it('does not reopen other terminal reply failures', async () => { + const h = bindingHarness({}, [ + exhaustedReply(0, 'deterministic_job_poison'), + ]); + await bindProviderMessageId(h.executor, input, { + stopContact: h.stopContact, + }); + expect(h.revive).not.toHaveBeenCalled(); + expect(h.stopContact).not.toHaveBeenCalled(); + }); + it('rejects an RFC identifier already owned by another job', async () => { + const h = bindingHarness({}, [], [{ id: 'another-job' }]); + await expect(bindProviderMessageId(h.executor, input)).rejects.toThrow( + /binding conflict/u + ); + expect(h.bind).not.toHaveBeenCalled(); + }); + it.each([ + 'provider-uuid', + '', + '', + ' ', + ])('rejects malformed RFC identifiers: %j', async (rfcMessageId) => { + const h = bindingHarness(); + await expect( + bindProviderMessageId(h.executor, { ...input, rfcMessageId }) + ).rejects.toThrow(/message_id/u); + expect(h.calls).toEqual([]); + }); + it('binds the actual provider ID without inventing a Gmail seed', async () => { + const h = bindingHarness(); + await bindProviderMessageId(h.executor, input, { + stopContact: h.stopContact, + }); + expect(h.bind.mock.calls[0]?.[0]).toEqual([ + jobId, + input.providerEmailId, + input.rfcMessageId, + ]); + }); + it('preserves an existing compatible Gmail seed and makes repeats idempotent', async () => { + const h = bindingHarness({ + rfc_message_id: input.rfcMessageId, + gmail_seed_message_id: 'legacy-gmail', + }); + await bindProviderMessageId(h.executor, input, { + stopContact: h.stopContact, + }); + expect(h.bind).not.toHaveBeenCalled(); + }); + it.each([ + { rfc_message_id: '' }, + { provider_email_id: 'different-provider' }, + { status: 'pending' }, + ])('rejects conflicting or unaccepted bindings: %j', async (row) => { + const h = bindingHarness(row); + await expect( + bindProviderMessageId(h.executor, input, { stopContact: h.stopContact }) + ).rejects.toThrow(); + expect(h.bind).not.toHaveBeenCalled(); + }); + it('settles a reply that arrived before the provider binding', async () => { + const h = bindingHarness({}, [ + { + id: 'reconcile-1', + status: 'pending', + contact_id: null, + payload: { + gmail_message_id: 'reply-gmail', + occurred_at: now.toISOString(), + in_reply_to: input.rfcMessageId, + references: [], + ranked_candidates: [{ message_id: input.rfcMessageId, rank: 0 }], + }, + }, + ]); + await bindProviderMessageId(h.executor, input, { + stopContact: h.stopContact, + }); + expect(h.stopContact).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ contactId, reason: 'campaign.reply_received' }) + ); + expect(h.complete).toHaveBeenCalledOnce(); + }); +}); diff --git a/libs/growth/src/lib/replies.ts b/libs/growth/src/lib/replies.ts index 0809c92a7..d8a993cfa 100644 --- a/libs/growth/src/lib/replies.ts +++ b/libs/growth/src/lib/replies.ts @@ -106,7 +106,9 @@ export interface ProcessGoogleMailboxEventInput { } export interface ProcessGoogleMailboxEventDependencies { - stopContact: typeof canonicalStopContact; + stopContact: ( + ...input: Parameters + ) => Promise; } export type GoogleMailboxRejectionReason = @@ -1092,12 +1094,23 @@ async function processSeed( if (updated.rows.length !== 1) domainError('seed_binding_conflict'); } + await reconcileBoundMessage(transaction, event, contactId, dependencies); + return 'seed_registered'; +} + +async function reconcileBoundMessage( + transaction: SqlTransaction, + event: { jobId: string; rfcMessageId: string }, + contactId: string, + dependencies: ProcessGoogleMailboxEventDependencies +): Promise { const pending = await transaction.execute( `/* growth:lock-google-reconcile-for-seed */ - select id, contact_id, status, payload + select id, contact_id, status, payload, last_error_code from growth_jobs where kind = 'reply_reconcile' - and status in ('pending', 'leased') + and (status in ('pending', 'leased') + or (status = 'failed' and last_error_code = 'founder_review')) and ( payload->>'in_reply_to' = $1 or payload->'references' ? $1 @@ -1111,6 +1124,37 @@ async function processSeed( ); for (const reconcile of pending.rows) { const rankedCandidates = rankedCandidatesFromPayload(reconcile.payload); + if (reconcile.status === 'failed') { + // A provider identity can arrive after the bounded reply lookup window. + // Recover only exhausted lookups with decisive direct-reply evidence. + // A lower reference remains ambiguous: roll back this binding so its + // contact cannot resume sending while that reply awaits resolution. + const direct = rankedCandidates + ? rankedCandidates.some( + (candidate) => + candidate.message_id === event.rfcMessageId && + candidate.rank === 0 + ) + : reconcile.payload['in_reply_to'] === event.rfcMessageId; + if (reconcile['last_error_code'] !== 'founder_review' || !direct) + domainError('reconcile_conflict'); + const occurredAt = reconcile.payload['occurred_at']; + if ( + typeof occurredAt !== 'string' || + Number.isNaN(new Date(occurredAt).getTime()) + ) + domainError('reconcile_payload_invalid'); + const revived = await transaction.execute<{ id: string }>( + `/* growth:revive-exhausted-google-reconcile */ + update growth_jobs set status = 'pending', available_at = $2, + lease_until = null, lease_token = null, last_error_code = null + where id = $1 and kind = 'reply_reconcile' + and status = 'failed' and last_error_code = 'founder_review' + returning id`, + [reconcile.id, new Date(occurredAt)] + ); + if (revived.rows.length !== 1) domainError('reconcile_conflict'); + } if (rankedCandidates) { const candidate = rankedCandidates.find( (item) => item.message_id === event.rfcMessageId @@ -1201,7 +1245,109 @@ async function processSeed( domainError('reconcile_conflict'); } } - return 'seed_registered'; +} + +export class ProviderMessageIdBindingError extends Error {} + +/** Bind only provider-authenticated identifiers, never a fabricated Gmail seed. */ +export async function bindProviderMessageId( + executor: SqlExecutor, + input: { providerEmailId: string; rfcMessageId: string }, + dependencies: ProcessGoogleMailboxEventDependencies = { + stopContact: canonicalStopContact, + } +): Promise<'bound' | 'unmatched' | 'ignored_deleted'> { + let rfcMessageId: string; + try { + rfcMessageId = parseMessageId('message_id', input.rfcMessageId); + } catch { + throw new ProviderMessageIdBindingError('Invalid provider message_id'); + } + if (!/^[A-Za-z0-9][A-Za-z0-9._:-]{0,255}$/u.test(input.providerEmailId)) { + throw new ProviderMessageIdBindingError('Invalid provider email ID'); + } + return executor.transaction(async (transaction) => { + // Same lock order as Gmail reconciliation: advisory, contact, then job. + await transaction.execute(`/* growth:acquire-google-reconcile-advisory-lock */ + select pg_advisory_xact_lock(hashtextextended('google-mailbox-reconciliation', 0))`); + const discovered = await transaction.execute( + `/* growth:discover-provider-message-job */ + select id, contact_id from growth_jobs where provider_email_id = $1`, + [input.providerEmailId] + ); + const reference = discovered.rows[0]; + if (!reference) return 'unmatched'; + if (!reference.contact_id) + throw new ProviderMessageIdBindingError( + 'Provider message contact is missing' + ); + const contact = await transaction.execute( + `/* growth:lock-provider-message-contact */ + select id, deleted_at from growth_contacts where id = $1 for update`, + [reference.contact_id] + ); + const locked = await transaction.execute( + `/* growth:lock-provider-message-job */ + select id, kind, contact_id, status, provider_email_id, delivery_status, + rfc_message_id, gmail_seed_message_id + from growth_jobs where id = $1 for update`, + [reference.id] + ); + const job = locked.rows[0]; + if ( + !contact.rows[0] || + !job || + job.id !== reference.id || + job.contact_id !== reference.contact_id || + job.provider_email_id !== input.providerEmailId || + job.status !== 'completed' || + !['fulfill', 'send_step'].includes(job.kind) || + !ACCEPTED_BOUND_DELIVERY_STATUSES.has(job.delivery_status) || + (job.rfc_message_id && job.rfc_message_id !== rfcMessageId) + ) { + throw new ProviderMessageIdBindingError( + 'Provider Message-ID binding conflict' + ); + } + if (contact.rows[0].deleted_at !== null) return 'ignored_deleted'; + const conflicts = await transaction.execute<{ id: string }>( + `/* growth:check-provider-message-conflicts */ + select id from growth_jobs where id <> $1 and rfc_message_id = $2 limit 1`, + [job.id, rfcMessageId] + ); + if (conflicts.rows.length) + throw new ProviderMessageIdBindingError( + 'Provider Message-ID binding conflict' + ); + if (!job.rfc_message_id) { + const updated = await transaction.execute<{ id: string }>( + `/* growth:bind-provider-message-id */ + update growth_jobs set rfc_message_id = $3 + where id = $1 and provider_email_id = $2 and rfc_message_id is null + returning id`, + [job.id, input.providerEmailId, rfcMessageId] + ); + if (updated.rows.length !== 1) + throw new ProviderMessageIdBindingError( + 'Provider Message-ID binding conflict' + ); + } + try { + await reconcileBoundMessage( + transaction, + { jobId: job.id, rfcMessageId }, + reference.contact_id, + dependencies + ); + } catch (error) { + if (error instanceof GoogleMailboxDomainError) + throw new ProviderMessageIdBindingError( + 'Provider reply reconciliation conflict' + ); + throw error; + } + return 'bound'; + }); } async function findReplyJob( diff --git a/libs/growth/src/lib/resend.spec.ts b/libs/growth/src/lib/resend.spec.ts index fcee95bd0..6cf01125f 100644 --- a/libs/growth/src/lib/resend.spec.ts +++ b/libs/growth/src/lib/resend.spec.ts @@ -12,6 +12,7 @@ import { APPROVED_ATTACHMENT_PATHS, RECIPIENT_EMAIL_SENDER, sendRecipientEmail, + reconcilePendingResendMessageIds, type RecipientDeliveryPolicy, type RecipientResendClient, } from './resend.ts'; @@ -179,7 +180,6 @@ describe('sendRecipientEmail', () => { { from: RECIPIENT_EMAIL_SENDER, to: 'developer@example.com', - bcc: RECIPIENT_EMAIL_SENDER, replyTo: RECIPIENT_EMAIL_SENDER, subject: message.subject, text: message.text, @@ -1001,7 +1001,7 @@ describe('sendRecipientEmail', () => { }); it.each(['preview', 'test'] as const)( - 'requires the production BCC mailbox on the %s allowlist before authorization', + 'allows %s recipient delivery without allowlisting an unused BCC mailbox', async (environment) => { const test = harness(); const policy = productionPolicy({ @@ -1013,9 +1013,8 @@ describe('sendRecipientEmail', () => { await expect( sendRecipientEmail(test.database, message, policy, test.dependencies) - ).rejects.toThrow(/bcc|allowlist/iu); - expect(test.authorizeLeasedJobForSubmission).not.toHaveBeenCalled(); - expect(test.send).not.toHaveBeenCalled(); + ).resolves.toEqual({ accepted: true, providerEmailId }); + expect(test.send.mock.calls[0]?.[0]).not.toHaveProperty('bcc'); } ); @@ -1145,3 +1144,153 @@ describe('sendRecipientEmail', () => { expect(test.send).not.toHaveBeenCalled(); }); }); + +describe('durable provider Message-ID lookup', () => { + function lookupHarness( + response: unknown = { + id: providerEmailId, + message_id: '', + tags: { environment: 'production' }, + } + ) { + const database = executor(); + const execute = vi + .fn() + .mockResolvedValue({ rows: [{ provider_email_id: providerEmailId }] }); + database.execute = execute; + const fetch = vi + .fn() + .mockResolvedValue(new Response(JSON.stringify(response))); + const bindProviderMessageId = vi.fn().mockResolvedValue('bound'); + return { + database, + execute, + fetch, + bindProviderMessageId, + dependencies: { + fetch, + bindProviderMessageId, + apiKey: 'test-key', + environment: 'production' as const, + databaseEnvironment: 'production' as const, + }, + }; + } + it('durably reserves a bounded retry before authenticated retrieval of the exact accepted ID', async () => { + const h = lookupHarness(); + await reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + h.dependencies + ); + expect(h.execute.mock.calls[0]?.[0]).toMatch( + /for update of job skip locked/iu + ); + expect(h.execute.mock.calls[0]?.[0]).toMatch(/message_id_lookup_after/u); + expect(h.execute.mock.calls[0]?.[0]).toMatch(/rfc_message_id is null/u); + expect(h.fetch).toHaveBeenCalledWith( + `https://api.resend.com/emails/${providerEmailId}`, + expect.objectContaining({ + method: 'GET', + headers: { Authorization: 'Bearer test-key' }, + redirect: 'error', + }) + ); + expect(h.bindProviderMessageId).toHaveBeenCalledWith(h.database, { + providerEmailId, + rfcMessageId: '', + }); + }); + it.each([ + { + id: 'different', + message_id: '', + tags: { environment: 'production' }, + }, + { + id: providerEmailId, + message_id: null, + tags: { environment: 'production' }, + }, + { + id: providerEmailId, + message_id: '', + tags: { environment: 'preview' }, + }, + ])( + 'leaves a durable retry for mismatched or missing identity: %j', + async (response) => { + const h = lookupHarness(response); + await reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + h.dependencies + ); + expect(h.bindProviderMessageId).not.toHaveBeenCalled(); + expect(h.execute).toHaveBeenCalledOnce(); + } + ); + it('retains the retry when retrieval fails, without submitting email', async () => { + const h = lookupHarness(); + h.fetch.mockRejectedValue(new Error('temporary')); + await expect( + reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + h.dependencies + ) + ).resolves.toEqual({ attempted: 1, bound: 0 }); + expect(h.bindProviderMessageId).not.toHaveBeenCalled(); + }); + it('accepts the provider GET tag array format', async () => { + const h = lookupHarness({ + id: providerEmailId, + message_id: '', + tags: [{ name: 'environment', value: 'production' }], + }); + await reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + h.dependencies + ); + expect(h.bindProviderMessageId).toHaveBeenCalledOnce(); + }); + it('propagates database failures during binding while retaining the reserved retry', async () => { + const h = lookupHarness(); + h.bindProviderMessageId.mockRejectedValue( + new Error('database unavailable') + ); + await expect( + reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + h.dependencies + ) + ).rejects.toThrow('database unavailable'); + expect(h.execute).toHaveBeenCalledOnce(); + }); + it('rejects a database environment mismatch before reserving or fetching', async () => { + const h = lookupHarness(); + await expect( + reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + { ...h.dependencies, databaseEnvironment: 'preview' } + ) + ).rejects.toThrow(/matching environments/u); + expect(h.execute).not.toHaveBeenCalled(); + expect(h.fetch).not.toHaveBeenCalled(); + }); + it('bounds the provider response and leaves a retry for oversized content', async () => { + const h = lookupHarness(); + h.fetch.mockResolvedValue(new Response('x'.repeat(131_073))); + await expect( + reconcilePendingResendMessageIds( + h.database, + { now, signal: new AbortController().signal }, + h.dependencies + ) + ).resolves.toEqual({ attempted: 1, bound: 0 }); + expect(h.bindProviderMessageId).not.toHaveBeenCalled(); + }); +}); diff --git a/libs/growth/src/lib/resend.ts b/libs/growth/src/lib/resend.ts index c9b14667e..f53dd5208 100644 --- a/libs/growth/src/lib/resend.ts +++ b/libs/growth/src/lib/resend.ts @@ -13,6 +13,10 @@ import { type UnsubscribeActionUrl, } from './tokens.ts'; import { normalizeRecipientEmail } from './crypto.ts'; +import { + bindProviderMessageId, + ProviderMessageIdBindingError, +} from './replies.ts'; export const RECIPIENT_EMAIL_SENDER = 'Brian at Threadplane '; @@ -21,7 +25,6 @@ const UUID_V4_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu; const OPAQUE_IDENTIFIER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]*$/u; const RECIPIENT_JOB_KINDS = new Set(['fulfill', 'send_step']); -const RECIPIENT_EMAIL_ADDRESS = 'brian@threadplane.ai'; const AMBIGUOUS_PROVIDER_ERROR_NAMES = new Set([ 'concurrent_idempotent_requests', 'invalid_idempotent_request', @@ -135,7 +138,6 @@ type ResendResponse = export interface RecipientEmailProviderPayload { from: typeof RECIPIENT_EMAIL_SENDER; to: string; - bcc: typeof RECIPIENT_EMAIL_SENDER; replyTo: typeof RECIPIENT_EMAIL_SENDER; subject: string; text: string; @@ -179,6 +181,7 @@ export type RecipientSendResult = | 'delivery_disabled' | 'outside_send_window' | 'mailbox_recovery_required' + | 'reply_binding_pending' | 'provider_rejected' | 'provider_outcome_unknown'; }; @@ -314,11 +317,6 @@ export function assertRecipientDeliveryPolicy( validEmail('nonProductionRecipientAllowlist', email) ) ); - if (!allowlist.has(RECIPIENT_EMAIL_ADDRESS)) { - throw new Error( - 'The recipient BCC mailbox must be on the non-production allowlist' - ); - } if (policy.nonProductionRedirectTo !== undefined) { const redirect = validEmail( 'nonProductionRedirectTo', @@ -503,7 +501,6 @@ export async function sendRecipientEmail( { from: RECIPIENT_EMAIL_SENDER, to, - bcc: RECIPIENT_EMAIL_SENDER, replyTo: RECIPIENT_EMAIL_SENDER, subject, text, @@ -595,3 +592,157 @@ export async function sendRecipientEmail( }); return { accepted: true, providerEmailId }; } + +export interface ResendMessageIdLookupDependencies { + apiKey: string; + environment: DeliveryEnvironment; + databaseEnvironment: DeliveryEnvironment; + fetch: typeof globalThis.fetch; + bindProviderMessageId: typeof bindProviderMessageId; +} + +function defaultMessageIdLookupDependencies(): ResendMessageIdLookupDependencies | null { + if ( + !process.env['RESEND_API_KEY']?.trim() || + !process.env['DELIVERY_ENVIRONMENT']?.trim() || + !process.env['GROWTH_DATABASE_ENVIRONMENT']?.trim() + ) + return null; + return { + apiKey: process.env['RESEND_API_KEY'] ?? '', + environment: process.env['DELIVERY_ENVIRONMENT'] as DeliveryEnvironment, + databaseEnvironment: process.env[ + 'GROWTH_DATABASE_ENVIRONMENT' + ] as DeliveryEnvironment, + fetch: globalThis.fetch, + bindProviderMessageId, + }; +} + +async function boundedProviderResponse( + response: Response +): Promise> { + if (!response.ok || !response.body) + throw new Error('Provider lookup unavailable'); + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let length = 0; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + length += value.byteLength; + if (length > 131_072) throw new Error('Provider lookup exceeds limit'); + chunks.push(value); + } + } finally { + await reader.cancel(); + } + const bytes = new Uint8Array(length); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + const value: unknown = JSON.parse(new TextDecoder().decode(bytes)); + if (!value || typeof value !== 'object' || Array.isArray(value)) + throw new Error('Invalid provider lookup'); + return value as Record; +} + +/** Accepted sends themselves are the durable retry queue; this path can only GET. */ +export async function reconcilePendingResendMessageIds( + executor: SqlExecutor, + input: { now: Date; signal: AbortSignal }, + dependencies: ResendMessageIdLookupDependencies | null = defaultMessageIdLookupDependencies() +): Promise<{ attempted: number; bound: number }> { + input.signal.throwIfAborted(); + // Non-mail dispatcher work remains available without delivery configuration. + // The final contact-specific send gate still requires every prior binding. + if (!dependencies) return { attempted: 0, bound: 0 }; + if ( + !['production', 'preview', 'test'].includes(dependencies.environment) || + dependencies.environment !== dependencies.databaseEnvironment || + !dependencies.apiKey.trim() || + Number.isNaN(input.now.getTime()) + ) { + throw new Error( + 'Provider lookup requires matching environments and credentials' + ); + } + // Reserve retries atomically before network I/O, including process crashes. + // A missing ID never reopens a completed send or changes its idempotency key. + const claimed = await executor.execute<{ provider_email_id: string }>( + `/* growth:claim-provider-message-lookups */ + with due as ( + select job.id from growth_jobs job + join growth_contacts contact on contact.id = job.contact_id + where job.kind in ('fulfill', 'send_step') and job.status = 'completed' + and job.provider_email_id is not null and job.rfc_message_id is null + and contact.deleted_at is null + and coalesce((job.payload->>'message_id_lookup_after')::timestamptz, job.updated_at) <= $1 + order by coalesce((job.payload->>'message_id_lookup_after')::timestamptz, job.updated_at), job.id + limit 5 for update of job skip locked + ) + update growth_jobs job set payload = job.payload || jsonb_build_object( + 'message_id_lookup_attempts', least(16, coalesce((job.payload->>'message_id_lookup_attempts')::int, 0) + 1), + 'message_id_lookup_after', $1::timestamptz + make_interval(mins => least(60, power(2, + least(6, coalesce((job.payload->>'message_id_lookup_attempts')::int, 0) + 1))::int)) + ) from due where job.id = due.id returning job.provider_email_id`, + [input.now] + ); + let bound = 0; + for (const row of claimed.rows) { + input.signal.throwIfAborted(); + let binding: { providerEmailId: string; rfcMessageId: string }; + try { + const providerEmailId = opaqueIdentifier( + 'providerEmailId', + row.provider_email_id, + 256 + ); + const response = await dependencies.fetch( + `https://api.resend.com/emails/${encodeURIComponent(providerEmailId)}`, + { + method: 'GET', + headers: { Authorization: `Bearer ${dependencies.apiKey}` }, + redirect: 'error', + signal: AbortSignal.any([input.signal, AbortSignal.timeout(5_000)]), + } + ); + const data = await boundedProviderResponse(response); + const tags = data['tags']; + const environment = Array.isArray(tags) + ? tags + .filter((tag) => tag && tag.name === 'environment') + .map((tag) => tag.value) + : tags && typeof tags === 'object' + ? [(tags as Record)['environment']] + : []; + if ( + data['id'] !== providerEmailId || + environment.length !== 1 || + environment[0] !== dependencies.environment || + typeof data['message_id'] !== 'string' + ) + continue; + binding = { providerEmailId, rfcMessageId: data['message_id'] }; + } catch { + input.signal.throwIfAborted(); + // The durable next-at reservation survives provider errors and timeouts. + continue; + } + try { + if ( + (await dependencies.bindProviderMessageId(executor, binding)) === + 'bound' + ) + bound += 1; + } catch (error) { + input.signal.throwIfAborted(); + if (!(error instanceof ProviderMessageIdBindingError)) throw error; + // Malformed/conflicting identities remain unbound; database errors surface. + } + } + return { attempted: claimed.rows.length, bound }; +} diff --git a/libs/growth/src/lib/webhooks.spec.ts b/libs/growth/src/lib/webhooks.spec.ts index 602ecf358..b75ddc92f 100644 --- a/libs/growth/src/lib/webhooks.spec.ts +++ b/libs/growth/src/lib/webhooks.spec.ts @@ -229,11 +229,31 @@ function webhookHarness( const dependencies: ProcessResendWebhookDependencies = { databaseEnvironment: 'production', stopContact, + bindProviderMessageId: vi.fn().mockResolvedValue('bound'), }; return { ...harness, stopContact, dependencies }; } describe('processVerifiedResendWebhook', () => { + it('binds an authenticated provider Message-ID using the exact provider email ID', async () => { + const h = webhookHarness(); + await processVerifiedResendWebhook( + h.executor, + { + providerEventId: 'msg_binding', + payload: event('email.sent', { message_id: '' }), + }, + h.dependencies + ); + expect(h.dependencies.bindProviderMessageId).toHaveBeenCalledWith( + expect.anything(), + { + providerEmailId, + rfcMessageId: '', + }, + { stopContact: h.stopContact } + ); + }); it('keeps supported parser fixtures assignable to the pinned Resend webhook union', () => { expect(supportedSdkFixtures).toHaveLength(7); }); @@ -779,6 +799,37 @@ describe('processVerifiedResendWebhook', () => { expect(harness.stopContact).not.toHaveBeenCalled(); }); + it('revisits binding reconciliation on duplicate provider events without duplicating delivery activity', async () => { + const h = webhookHarness({ + existingActivity: { + event_key: 'resend:msg_duplicate_binding', + contact_id: contactId, + project_id: null, + kind: 'delivery.sent', + occurred_at: now, + data: { + provider: 'resend', + provider_event_id: 'msg_duplicate_binding', + provider_email_id: providerEmailId, + event_type: 'email.sent', + category: 'sent', + }, + }, + }); + await expect( + processVerifiedResendWebhook( + h.executor, + { + providerEventId: 'msg_duplicate_binding', + payload: event('email.sent', { message_id: '' }), + }, + h.dependencies + ) + ).resolves.toEqual({ applied: false, reason: 'replay' }); + expect(h.dependencies.bindProviderMessageId).toHaveBeenCalledOnce(); + expect(h.calls).not.toContain('insert-resend-webhook-activity'); + }); + it('fails conflicting reuse of a provider event ID before status mutation', async () => { const harness = webhookHarness({ existingActivity: { diff --git a/libs/growth/src/lib/webhooks.ts b/libs/growth/src/lib/webhooks.ts index ab64d16da..fd93d03ce 100644 --- a/libs/growth/src/lib/webhooks.ts +++ b/libs/growth/src/lib/webhooks.ts @@ -1,6 +1,7 @@ import type { SqlExecutor, SqlTransaction } from './database.ts'; import type { GrowthDeliveryStatus } from './models.ts'; import { isCampaignTemplateId, type DeliveryEnvironment } from './resend.ts'; +import { bindProviderMessageId } from './replies.ts'; import { stopContact, type CanonicalStopReason, @@ -46,6 +47,7 @@ interface ParsedResendEvent { type: SupportedResendEventType; occurredAt: Date; providerEmailId: string; + rfcMessageId?: string; tags: Record; bounceCategory?: 'permanent' | 'transient' | 'unknown'; } @@ -72,6 +74,7 @@ interface WebhookActivityRow extends Record { export interface ProcessResendWebhookDependencies { databaseEnvironment: DeliveryEnvironment; + bindProviderMessageId?: typeof bindProviderMessageId; stopContact: ( executor: SqlExecutor, input: StopContactInput @@ -272,6 +275,9 @@ function parseSupportedEvent(payload: unknown): ParsedResendEvent | null { type, occurredAt, providerEmailId, + ...(data['message_id'] == null + ? {} + : { rfcMessageId: boundedText(data['message_id'], 998) }), tags, ...(type === 'email.bounced' ? { bounceCategory: validateClosedDetails(type, data) } @@ -508,6 +514,21 @@ export async function processVerifiedResendWebhook( }; return executor.transaction(async (transaction) => { + // Serialize with Gmail before acquiring contact/job locks. Replays also + // revisit reconciliation so a late reply cannot miss an existing binding. + if ( + event.rfcMessageId && + ['fulfill', 'send_step'].includes(event.tags['job_kind'] ?? '') + ) { + await (dependencies.bindProviderMessageId ?? bindProviderMessageId)( + transactionExecutor(transaction), + { + providerEmailId: event.providerEmailId, + rfcMessageId: event.rfcMessageId, + }, + { stopContact: dependencies.stopContact } + ); + } const existing = await transaction.execute( `/* growth:read-resend-webhook-activity */ select event_key, contact_id, project_id, kind, occurred_at, data diff --git a/libs/growth/test/forms.integration.spec.ts b/libs/growth/test/forms.integration.spec.ts index 506333e95..6c21933eb 100644 --- a/libs/growth/test/forms.integration.spec.ts +++ b/libs/growth/test/forms.integration.spec.ts @@ -121,6 +121,82 @@ describeDatabase( return result.rows[0]; } + it('retains blocked forms without mail jobs and cannot overwrite a legitimate identity', async () => { + const email = `form-abuse-${randomUUID()}@example.com`; + const now = new Date(); + try { + const legitimate = await acceptFormSubmission(executor, { + ...submission(email, randomUUID(), 'chat', now), + displayName: 'Ada Lovelace', + companyName: 'Analytical Engines', + }); + const junkId = randomUUID(); + const input: AcceptFormSubmissionInput = { + ...submission(email, junkId, 'chat', now), + displayName: 'aBcDeFgHiJkLmNoP', + companyName: 'Zqxwy LLC', + form: { kind: 'contact', message: 'qRsTuVwXyZaBcDeF' }, + sourceForm: 'contact', + }; + const junk = await acceptFormSubmission(executor, input); + expect(junk).toMatchObject({ + approved: false, + deliverySuppressed: true, + contactId: legitimate.contactId, + }); + expect((await counts(email, junkId)).jobs).toBe('0'); + expect( + await acceptFormSubmission(executor, { + ...input, + occurredAt: new Date(now.getTime() + 1000), + }) + ).toEqual(junk); + const contact = await executor.execute<{ + display_name: string; + company_name: string; + outreach_approved_at: Date; + }>( + 'select display_name,company_name,outreach_approved_at from growth_contacts where id=$1', + [legitimate.contactId] + ); + expect(contact.rows[0]).toMatchObject({ + display_name: 'Ada Lovelace', + company_name: 'Analytical Engines', + }); + expect(contact.rows[0].outreach_approved_at).not.toBeNull(); + } finally { + await cleanup(email); + } + }); + + it('records a new honeypot contact without approval, fulfillment, or internal notification', async () => { + const email = `honeypot-${randomUUID()}@example.com`; + const id = randomUUID(); + try { + const result = await acceptFormSubmission(executor, { + ...submission(email, id, 'chat', new Date()), + honeypot: 'https://trap.invalid', + }); + expect(result).toMatchObject({ + approved: false, + deliverySuppressed: true, + }); + expect((await counts(email, id)).jobs).toBe('0'); + const assessment = await executor.execute<{ + data: Record; + }>('select data from growth_activity where event_key=$1', [ + `form:${id}:assessment`, + ]); + expect(assessment.rows[0].data['assessment']).toMatchObject({ + category: 'honeypot', + score: 100, + decision: 'blocked', + }); + } finally { + await cleanup(email); + } + }); + async function collisionCounts( emails: readonly string[], submissionId: string