Set up email infrastructure
X-Lovable-Edit-ID: edt-7046418a-4423-4e5e-8020-031496b9aa9b Co-authored-by: renee-png <262607627+renee-png@users.noreply.github.com>
This commit is contained in:
gpt-engineer-app[bot]
+3
-1
@@ -17,6 +17,8 @@
|
||||
"@dnd-kit/sortable": "^10.0.0",
|
||||
"@dnd-kit/utilities": "^3.2.2",
|
||||
"@hookform/resolvers": "3.10.0",
|
||||
"@lovable.dev/email-js": "^0.0.4",
|
||||
"@lovable.dev/webhooks-js": "^0.0.1",
|
||||
"@radix-ui/react-accordion": "^1.2.12",
|
||||
"@radix-ui/react-alert-dialog": "^1.1.15",
|
||||
"@radix-ui/react-aspect-ratio": "^1.1.8",
|
||||
@@ -43,7 +45,7 @@
|
||||
"@radix-ui/react-toggle": "^1.1.10",
|
||||
"@radix-ui/react-toggle-group": "^1.1.11",
|
||||
"@radix-ui/react-tooltip": "^1.2.8",
|
||||
"@supabase/supabase-js": "^2.103.3",
|
||||
"@supabase/supabase-js": "^2.104.1",
|
||||
"@tailwindcss/vite": "^4.2.1",
|
||||
"@tanstack/react-query": "^5.83.0",
|
||||
"@tanstack/react-router": "^1.168.0",
|
||||
|
||||
+22
-9
@@ -57,6 +57,7 @@ import { Route as DocumentsTemplatesNewRouteImport } from './routes/documents.te
|
||||
import { Route as DocumentsTemplatesTemplateIdRouteImport } from './routes/documents.templates.$templateId'
|
||||
import { Route as DocumentsPleadingNewRouteImport } from './routes/documents.pleading.new'
|
||||
import { Route as ClientsClientIdFeesRouteImport } from './routes/clients.$clientId.fees'
|
||||
import { Route as LovableEmailQueueProcessRouteImport } from './routes/lovable/email/queue/process'
|
||||
|
||||
const SetupRoute = SetupRouteImport.update({
|
||||
id: '/setup',
|
||||
@@ -299,6 +300,12 @@ const ClientsClientIdFeesRoute = ClientsClientIdFeesRouteImport.update({
|
||||
path: '/fees',
|
||||
getParentRoute: () => ClientsClientIdRoute,
|
||||
} as any)
|
||||
const LovableEmailQueueProcessRoute =
|
||||
LovableEmailQueueProcessRouteImport.update({
|
||||
id: '/lovable/email/queue/process',
|
||||
path: '/lovable/email/queue/process',
|
||||
getParentRoute: () => rootRouteImport,
|
||||
} as any)
|
||||
|
||||
export interface FileRoutesByFullPath {
|
||||
'/': typeof IndexRoute
|
||||
@@ -349,6 +356,7 @@ export interface FileRoutesByFullPath {
|
||||
'/documents/templates/new': typeof DocumentsTemplatesNewRoute
|
||||
'/invoices/new/$clientId': typeof InvoicesNewClientIdRoute
|
||||
'/documents/templates/': typeof DocumentsTemplatesIndexRoute
|
||||
'/lovable/email/queue/process': typeof LovableEmailQueueProcessRoute
|
||||
}
|
||||
export interface FileRoutesByTo {
|
||||
'/': typeof IndexRoute
|
||||
@@ -398,6 +406,7 @@ export interface FileRoutesByTo {
|
||||
'/documents/templates/new': typeof DocumentsTemplatesNewRoute
|
||||
'/invoices/new/$clientId': typeof InvoicesNewClientIdRoute
|
||||
'/documents/templates': typeof DocumentsTemplatesIndexRoute
|
||||
'/lovable/email/queue/process': typeof LovableEmailQueueProcessRoute
|
||||
}
|
||||
export interface FileRoutesById {
|
||||
__root__: typeof rootRouteImport
|
||||
@@ -449,6 +458,7 @@ export interface FileRoutesById {
|
||||
'/documents/templates/new': typeof DocumentsTemplatesNewRoute
|
||||
'/invoices/new/$clientId': typeof InvoicesNewClientIdRoute
|
||||
'/documents/templates/': typeof DocumentsTemplatesIndexRoute
|
||||
'/lovable/email/queue/process': typeof LovableEmailQueueProcessRoute
|
||||
}
|
||||
export interface FileRouteTypes {
|
||||
fileRoutesByFullPath: FileRoutesByFullPath
|
||||
@@ -501,6 +511,7 @@ export interface FileRouteTypes {
|
||||
| '/documents/templates/new'
|
||||
| '/invoices/new/$clientId'
|
||||
| '/documents/templates/'
|
||||
| '/lovable/email/queue/process'
|
||||
fileRoutesByTo: FileRoutesByTo
|
||||
to:
|
||||
| '/'
|
||||
@@ -550,6 +561,7 @@ export interface FileRouteTypes {
|
||||
| '/documents/templates/new'
|
||||
| '/invoices/new/$clientId'
|
||||
| '/documents/templates'
|
||||
| '/lovable/email/queue/process'
|
||||
id:
|
||||
| '__root__'
|
||||
| '/'
|
||||
@@ -600,6 +612,7 @@ export interface FileRouteTypes {
|
||||
| '/documents/templates/new'
|
||||
| '/invoices/new/$clientId'
|
||||
| '/documents/templates/'
|
||||
| '/lovable/email/queue/process'
|
||||
fileRoutesById: FileRoutesById
|
||||
}
|
||||
export interface RootRouteChildren {
|
||||
@@ -638,6 +651,7 @@ export interface RootRouteChildren {
|
||||
DocumentsTemplatesTemplateIdRoute: typeof DocumentsTemplatesTemplateIdRoute
|
||||
DocumentsTemplatesNewRoute: typeof DocumentsTemplatesNewRoute
|
||||
DocumentsTemplatesIndexRoute: typeof DocumentsTemplatesIndexRoute
|
||||
LovableEmailQueueProcessRoute: typeof LovableEmailQueueProcessRoute
|
||||
}
|
||||
|
||||
declare module '@tanstack/react-router' {
|
||||
@@ -978,6 +992,13 @@ declare module '@tanstack/react-router' {
|
||||
preLoaderRoute: typeof ClientsClientIdFeesRouteImport
|
||||
parentRoute: typeof ClientsClientIdRoute
|
||||
}
|
||||
'/lovable/email/queue/process': {
|
||||
id: '/lovable/email/queue/process'
|
||||
path: '/lovable/email/queue/process'
|
||||
fullPath: '/lovable/email/queue/process'
|
||||
preLoaderRoute: typeof LovableEmailQueueProcessRouteImport
|
||||
parentRoute: typeof rootRouteImport
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1073,16 +1094,8 @@ const rootRouteChildren: RootRouteChildren = {
|
||||
DocumentsTemplatesTemplateIdRoute: DocumentsTemplatesTemplateIdRoute,
|
||||
DocumentsTemplatesNewRoute: DocumentsTemplatesNewRoute,
|
||||
DocumentsTemplatesIndexRoute: DocumentsTemplatesIndexRoute,
|
||||
LovableEmailQueueProcessRoute: LovableEmailQueueProcessRoute,
|
||||
}
|
||||
export const routeTree = rootRouteImport
|
||||
._addFileChildren(rootRouteChildren)
|
||||
._addFileTypes<FileRouteTypes>()
|
||||
|
||||
import type { getRouter } from './router.tsx'
|
||||
import type { createStart } from '@tanstack/react-start'
|
||||
declare module '@tanstack/react-start' {
|
||||
interface Register {
|
||||
ssr: true
|
||||
router: Awaited<ReturnType<typeof getRouter>>
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,331 @@
|
||||
import { sendLovableEmail } from '@lovable.dev/email-js'
|
||||
import { createClient } from '@supabase/supabase-js'
|
||||
import { createFileRoute } from '@tanstack/react-router'
|
||||
|
||||
// The email infra tables (email_send_log, email_send_state, etc.) are not in
|
||||
// the generated Database types, so we use an untyped client throughout this
|
||||
// route. Tables are created by the email infra migration.
|
||||
type AnyClient = ReturnType<typeof createClient<any, any, any>>
|
||||
|
||||
const MAX_RETRIES = 5
|
||||
const DEFAULT_BATCH_SIZE = 10
|
||||
const DEFAULT_SEND_DELAY_MS = 200
|
||||
const DEFAULT_AUTH_TTL_MINUTES = 15
|
||||
const DEFAULT_TRANSACTIONAL_TTL_MINUTES = 60
|
||||
|
||||
// Check if an error is a rate-limit (429) response.
|
||||
// Uses EmailAPIError.status when available (email-js >=0.x with structured errors),
|
||||
// falls back to parsing the error message for older versions.
|
||||
function isRateLimited(error: unknown): boolean {
|
||||
if (error && typeof error === 'object' && 'status' in error) {
|
||||
return (error as { status: number }).status === 429
|
||||
}
|
||||
return error instanceof Error && error.message.includes('429')
|
||||
}
|
||||
|
||||
// Check if an error is a forbidden (403) response, which means emails are
|
||||
// disabled for this project. Retrying won't help — move straight to DLQ.
|
||||
function isForbidden(error: unknown): boolean {
|
||||
if (error && typeof error === 'object' && 'status' in error) {
|
||||
return (error as { status: number }).status === 403
|
||||
}
|
||||
return error instanceof Error && error.message.includes('403')
|
||||
}
|
||||
|
||||
// Extract Retry-After seconds from a structured EmailAPIError, or default to 60s.
|
||||
function getRetryAfterSeconds(error: unknown): number {
|
||||
if (error && typeof error === 'object' && 'retryAfterSeconds' in error) {
|
||||
return (error as { retryAfterSeconds: number | null }).retryAfterSeconds ?? 60
|
||||
}
|
||||
return 60
|
||||
}
|
||||
|
||||
// Move a message to the dead letter queue and log the reason.
|
||||
async function moveToDlq(
|
||||
supabase: AnyClient,
|
||||
queue: string,
|
||||
msg: { msg_id: number; message: Record<string, unknown> },
|
||||
reason: string
|
||||
): Promise<void> {
|
||||
const payload = msg.message
|
||||
await supabase.from('email_send_log').insert({
|
||||
message_id: payload.message_id,
|
||||
template_name: (payload.label || queue) as string,
|
||||
recipient_email: payload.to,
|
||||
status: 'dlq',
|
||||
error_message: reason,
|
||||
})
|
||||
const { error } = await supabase.rpc('move_to_dlq', {
|
||||
source_queue: queue,
|
||||
dlq_name: `${queue}_dlq`,
|
||||
message_id: msg.msg_id,
|
||||
payload,
|
||||
})
|
||||
if (error) {
|
||||
console.error('Failed to move message to DLQ', { queue, msg_id: msg.msg_id, reason, error })
|
||||
}
|
||||
}
|
||||
|
||||
export const Route = createFileRoute("/lovable/email/queue/process")({
|
||||
server: {
|
||||
handlers: {
|
||||
POST: async ({ request }) => {
|
||||
const apiKey = process.env.LOVABLE_API_KEY
|
||||
const supabaseUrl = import.meta.env.VITE_SUPABASE_URL
|
||||
const supabaseServiceKey = process.env.SUPABASE_SERVICE_ROLE_KEY
|
||||
|
||||
if (!apiKey || !supabaseUrl || !supabaseServiceKey) {
|
||||
console.error('Missing required environment variables')
|
||||
return Response.json(
|
||||
{ error: 'Server configuration error' },
|
||||
{ status: 500 }
|
||||
)
|
||||
}
|
||||
|
||||
// Verify the caller is authorized with the service role key.
|
||||
// In the TanStack stack, the pg_cron job sends the service role key as a Bearer token.
|
||||
const authHeader = request.headers.get('Authorization')
|
||||
if (!authHeader?.startsWith('Bearer ')) {
|
||||
return Response.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
const token = authHeader.slice('Bearer '.length).trim()
|
||||
if (token !== supabaseServiceKey) {
|
||||
return Response.json({ error: 'Forbidden' }, { status: 403 })
|
||||
}
|
||||
|
||||
const supabase = createClient<any, any, any>(supabaseUrl, supabaseServiceKey)
|
||||
|
||||
// 1. Check rate-limit cooldown and read queue config
|
||||
const { data: state } = await supabase
|
||||
.from('email_send_state')
|
||||
.select('retry_after_until, batch_size, send_delay_ms, auth_email_ttl_minutes, transactional_email_ttl_minutes')
|
||||
.single()
|
||||
|
||||
if (state?.retry_after_until && new Date(state.retry_after_until) > new Date()) {
|
||||
return Response.json({ skipped: true, reason: 'rate_limited' })
|
||||
}
|
||||
|
||||
const batchSize = state?.batch_size ?? DEFAULT_BATCH_SIZE
|
||||
const sendDelayMs = state?.send_delay_ms ?? DEFAULT_SEND_DELAY_MS
|
||||
const ttlMinutes: Record<string, number> = {
|
||||
auth_emails: state?.auth_email_ttl_minutes ?? DEFAULT_AUTH_TTL_MINUTES,
|
||||
transactional_emails: state?.transactional_email_ttl_minutes ?? DEFAULT_TRANSACTIONAL_TTL_MINUTES,
|
||||
}
|
||||
|
||||
let totalProcessed = 0
|
||||
|
||||
// 2. Process auth_emails first (priority), then transactional_emails
|
||||
for (const queue of ['auth_emails', 'transactional_emails']) {
|
||||
const { data: messages, error: readError } = await supabase.rpc('read_email_batch', {
|
||||
queue_name: queue,
|
||||
batch_size: batchSize,
|
||||
vt: 30,
|
||||
})
|
||||
|
||||
if (readError) {
|
||||
console.error('Failed to read email batch', { queue, error: readError })
|
||||
continue
|
||||
}
|
||||
|
||||
if (!messages?.length) continue
|
||||
|
||||
// Retry budget is based on real send failures, not pgmq read_ct.
|
||||
const messageIds = Array.from(
|
||||
new Set(
|
||||
messages
|
||||
.map((msg: any) =>
|
||||
msg?.message?.message_id && typeof msg.message.message_id === 'string'
|
||||
? msg.message.message_id
|
||||
: null
|
||||
)
|
||||
.filter((id: string | null): id is string => Boolean(id))
|
||||
)
|
||||
)
|
||||
const failedAttemptsByMessageId = new Map<string, number>()
|
||||
if (messageIds.length > 0) {
|
||||
const { data: failedRows, error: failedRowsError } = await supabase
|
||||
.from('email_send_log')
|
||||
.select('message_id')
|
||||
.in('message_id', messageIds)
|
||||
.eq('status', 'failed')
|
||||
|
||||
if (failedRowsError) {
|
||||
console.error('Failed to load failed-attempt counters', {
|
||||
queue,
|
||||
error: failedRowsError,
|
||||
})
|
||||
} else {
|
||||
for (const row of failedRows ?? []) {
|
||||
const messageId = row?.message_id
|
||||
if (typeof messageId !== 'string' || !messageId) continue
|
||||
failedAttemptsByMessageId.set(
|
||||
messageId,
|
||||
(failedAttemptsByMessageId.get(messageId) ?? 0) + 1
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (let i = 0; i < messages.length; i++) {
|
||||
const msg = messages[i]
|
||||
const payload = msg.message
|
||||
const failedAttempts =
|
||||
payload?.message_id && typeof payload.message_id === 'string'
|
||||
? (failedAttemptsByMessageId.get(payload.message_id) ?? 0)
|
||||
: msg.read_ct ?? 0
|
||||
|
||||
// Drop expired messages (TTL exceeded).
|
||||
// Prefer payload.queued_at when present; fall back to PGMQ's enqueued_at
|
||||
// which is always set by the queue.
|
||||
const queuedAt = payload.queued_at ?? msg.enqueued_at
|
||||
if (queuedAt) {
|
||||
const ageMs = Date.now() - new Date(queuedAt).getTime()
|
||||
const maxAgeMs = ttlMinutes[queue] * 60 * 1000
|
||||
if (ageMs > maxAgeMs) {
|
||||
console.warn('Email expired (TTL exceeded)', {
|
||||
queue,
|
||||
msg_id: msg.msg_id,
|
||||
queued_at: queuedAt,
|
||||
ttl_minutes: ttlMinutes[queue],
|
||||
})
|
||||
await moveToDlq(supabase, queue, msg, `TTL exceeded (${ttlMinutes[queue]} minutes)`)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// Move to DLQ if max failed send attempts reached.
|
||||
if (failedAttempts >= MAX_RETRIES) {
|
||||
await moveToDlq(supabase, queue, msg, `Max retries (${MAX_RETRIES}) exceeded (attempted ${failedAttempts} times)`)
|
||||
continue
|
||||
}
|
||||
|
||||
// Guard: skip if another worker already sent this message (VT expired race)
|
||||
if (payload.message_id) {
|
||||
const { data: alreadySent } = await supabase
|
||||
.from('email_send_log')
|
||||
.select('id')
|
||||
.eq('message_id', payload.message_id)
|
||||
.eq('status', 'sent')
|
||||
.maybeSingle()
|
||||
|
||||
if (alreadySent) {
|
||||
console.warn('Skipping duplicate send (already sent)', {
|
||||
queue,
|
||||
msg_id: msg.msg_id,
|
||||
message_id: payload.message_id,
|
||||
})
|
||||
const { error: dupDelError } = await supabase.rpc('delete_email', {
|
||||
queue_name: queue,
|
||||
message_id: msg.msg_id,
|
||||
})
|
||||
if (dupDelError) {
|
||||
console.error('Failed to delete duplicate message from queue', { queue, msg_id: msg.msg_id, error: dupDelError })
|
||||
}
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
await sendLovableEmail(
|
||||
{
|
||||
run_id: payload.run_id,
|
||||
to: payload.to,
|
||||
from: payload.from,
|
||||
sender_domain: payload.sender_domain,
|
||||
subject: payload.subject,
|
||||
html: payload.html,
|
||||
text: payload.text,
|
||||
purpose: payload.purpose,
|
||||
label: payload.label,
|
||||
idempotency_key: payload.idempotency_key,
|
||||
unsubscribe_token: payload.unsubscribe_token,
|
||||
message_id: payload.message_id,
|
||||
},
|
||||
{ apiKey, sendUrl: process.env.LOVABLE_SEND_URL }
|
||||
)
|
||||
|
||||
// Log success
|
||||
await supabase.from('email_send_log').insert({
|
||||
message_id: payload.message_id,
|
||||
template_name: payload.label || queue,
|
||||
recipient_email: payload.to,
|
||||
status: 'sent',
|
||||
})
|
||||
|
||||
// Delete from queue
|
||||
const { error: delError } = await supabase.rpc('delete_email', {
|
||||
queue_name: queue,
|
||||
message_id: msg.msg_id,
|
||||
})
|
||||
if (delError) {
|
||||
console.error('Failed to delete sent message from queue', { queue, msg_id: msg.msg_id, error: delError })
|
||||
}
|
||||
totalProcessed++
|
||||
} catch (error) {
|
||||
const errorMsg = error instanceof Error ? error.message : String(error)
|
||||
console.error('Email send failed', {
|
||||
queue,
|
||||
msg_id: msg.msg_id,
|
||||
read_ct: msg.read_ct,
|
||||
failed_attempts: failedAttempts,
|
||||
error: errorMsg,
|
||||
})
|
||||
|
||||
if (isRateLimited(error)) {
|
||||
await supabase.from('email_send_log').insert({
|
||||
message_id: payload.message_id,
|
||||
template_name: payload.label || queue,
|
||||
recipient_email: payload.to,
|
||||
status: 'failed',
|
||||
error_message: errorMsg.slice(0, 1000),
|
||||
})
|
||||
|
||||
const retryAfterSecs = getRetryAfterSeconds(error)
|
||||
await supabase
|
||||
.from('email_send_state')
|
||||
.update({
|
||||
retry_after_until: new Date(
|
||||
Date.now() + retryAfterSecs * 1000
|
||||
).toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
})
|
||||
.eq('id', 1)
|
||||
|
||||
// Stop processing — remaining messages stay in queue (VT expires, retried next cycle)
|
||||
return Response.json({ processed: totalProcessed, stopped: 'rate_limited' })
|
||||
}
|
||||
|
||||
// 403 means emails are disabled for this project — retrying won't help.
|
||||
if (isForbidden(error)) {
|
||||
await moveToDlq(supabase, queue, msg, 'Emails disabled for this project')
|
||||
return Response.json({ processed: totalProcessed, stopped: 'emails_disabled' })
|
||||
}
|
||||
|
||||
// Log non-429 failures to track real retry attempts.
|
||||
await supabase.from('email_send_log').insert({
|
||||
message_id: payload.message_id,
|
||||
template_name: payload.label || queue,
|
||||
recipient_email: payload.to,
|
||||
status: 'failed',
|
||||
error_message: errorMsg.slice(0, 1000),
|
||||
})
|
||||
if (payload?.message_id && typeof payload.message_id === 'string') {
|
||||
failedAttemptsByMessageId.set(payload.message_id, failedAttempts + 1)
|
||||
}
|
||||
|
||||
// Non-429 errors: message stays invisible until VT expires, then retried
|
||||
}
|
||||
|
||||
// Small delay between sends to smooth bursts
|
||||
if (i < messages.length - 1) {
|
||||
await new Promise((r) => setTimeout(r, sendDelayMs))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return Response.json({ processed: totalProcessed })
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
@@ -0,0 +1,292 @@
|
||||
-- Email infrastructure
|
||||
-- Creates the queue system, send log, send state, suppression, and unsubscribe
|
||||
-- tables used by both auth and transactional emails.
|
||||
|
||||
-- Extensions required for queue processing
|
||||
CREATE EXTENSION IF NOT EXISTS pg_net SCHEMA extensions;
|
||||
DO $$ BEGIN
|
||||
IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_cron') THEN
|
||||
CREATE EXTENSION pg_cron;
|
||||
END IF;
|
||||
END $$;
|
||||
CREATE EXTENSION IF NOT EXISTS supabase_vault;
|
||||
CREATE EXTENSION IF NOT EXISTS pgmq;
|
||||
|
||||
-- Create email queues (auth = high priority, transactional = normal)
|
||||
-- Wrapped in DO blocks to handle "queue already exists" errors idempotently.
|
||||
DO $$ BEGIN PERFORM pgmq.create('auth_emails'); EXCEPTION WHEN OTHERS THEN NULL; END $$;
|
||||
DO $$ BEGIN PERFORM pgmq.create('transactional_emails'); EXCEPTION WHEN OTHERS THEN NULL; END $$;
|
||||
|
||||
-- Dead-letter queues for messages that exceed max retries
|
||||
DO $$ BEGIN PERFORM pgmq.create('auth_emails_dlq'); EXCEPTION WHEN OTHERS THEN NULL; END $$;
|
||||
DO $$ BEGIN PERFORM pgmq.create('transactional_emails_dlq'); EXCEPTION WHEN OTHERS THEN NULL; END $$;
|
||||
|
||||
-- Email send log table (audit trail for all send attempts)
|
||||
-- UPDATE is allowed for the service role so the suppression edge function
|
||||
-- can update a log record's status when a bounce/complaint/unsubscribe occurs.
|
||||
CREATE TABLE IF NOT EXISTS public.email_send_log (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
message_id TEXT,
|
||||
template_name TEXT NOT NULL,
|
||||
recipient_email TEXT NOT NULL,
|
||||
status TEXT NOT NULL CHECK (status IN ('pending', 'sent', 'suppressed', 'failed', 'bounced', 'complained', 'dlq')),
|
||||
error_message TEXT,
|
||||
metadata JSONB,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
ALTER TABLE public.email_send_log ENABLE ROW LEVEL SECURITY;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can read send log"
|
||||
ON public.email_send_log FOR SELECT
|
||||
USING (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can insert send log"
|
||||
ON public.email_send_log FOR INSERT
|
||||
WITH CHECK (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can update send log"
|
||||
ON public.email_send_log FOR UPDATE
|
||||
USING (auth.role() = 'service_role')
|
||||
WITH CHECK (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_email_send_log_created ON public.email_send_log(created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_email_send_log_recipient ON public.email_send_log(recipient_email);
|
||||
|
||||
-- Backfill: add message_id column to existing tables that predate this migration
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE public.email_send_log ADD COLUMN message_id TEXT;
|
||||
EXCEPTION WHEN duplicate_column THEN NULL;
|
||||
END $$;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_email_send_log_message ON public.email_send_log(message_id);
|
||||
|
||||
-- Prevent duplicate sends: only one 'sent' row per message_id.
|
||||
-- If VT expires and another worker picks up the same message, the pre-send
|
||||
-- check catches it. This index is a DB-level safety net for race conditions.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_email_send_log_message_sent_unique
|
||||
ON public.email_send_log(message_id) WHERE status = 'sent';
|
||||
|
||||
-- Backfill: update status CHECK constraint for existing tables that predate new statuses
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE public.email_send_log DROP CONSTRAINT IF EXISTS email_send_log_status_check;
|
||||
ALTER TABLE public.email_send_log ADD CONSTRAINT email_send_log_status_check
|
||||
CHECK (status IN ('pending', 'sent', 'suppressed', 'failed', 'bounced', 'complained', 'dlq'));
|
||||
END $$;
|
||||
|
||||
-- Rate-limit state and queue config (single row, tracks Retry-After cooldown + throughput settings)
|
||||
CREATE TABLE IF NOT EXISTS public.email_send_state (
|
||||
id INT PRIMARY KEY DEFAULT 1 CHECK (id = 1),
|
||||
retry_after_until TIMESTAMPTZ,
|
||||
batch_size INTEGER NOT NULL DEFAULT 10,
|
||||
send_delay_ms INTEGER NOT NULL DEFAULT 200,
|
||||
auth_email_ttl_minutes INTEGER NOT NULL DEFAULT 15,
|
||||
transactional_email_ttl_minutes INTEGER NOT NULL DEFAULT 60,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
INSERT INTO public.email_send_state (id) VALUES (1) ON CONFLICT DO NOTHING;
|
||||
|
||||
-- Backfill: add config columns to existing tables that predate this migration
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE public.email_send_state ADD COLUMN batch_size INTEGER NOT NULL DEFAULT 10;
|
||||
EXCEPTION WHEN duplicate_column THEN NULL;
|
||||
END $$;
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE public.email_send_state ADD COLUMN send_delay_ms INTEGER NOT NULL DEFAULT 200;
|
||||
EXCEPTION WHEN duplicate_column THEN NULL;
|
||||
END $$;
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE public.email_send_state ADD COLUMN auth_email_ttl_minutes INTEGER NOT NULL DEFAULT 15;
|
||||
EXCEPTION WHEN duplicate_column THEN NULL;
|
||||
END $$;
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE public.email_send_state ADD COLUMN transactional_email_ttl_minutes INTEGER NOT NULL DEFAULT 60;
|
||||
EXCEPTION WHEN duplicate_column THEN NULL;
|
||||
END $$;
|
||||
|
||||
ALTER TABLE public.email_send_state ENABLE ROW LEVEL SECURITY;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can manage send state"
|
||||
ON public.email_send_state FOR ALL
|
||||
USING (auth.role() = 'service_role')
|
||||
WITH CHECK (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
-- RPC wrappers so Edge Functions can interact with pgmq via supabase.rpc()
|
||||
-- (PostgREST only exposes functions in the public schema; pgmq functions are in the pgmq schema)
|
||||
-- All wrappers auto-create the queue on undefined_table (42P01) so emails
|
||||
-- are never lost if the queue was dropped (extension upgrade, restore, etc.).
|
||||
CREATE OR REPLACE FUNCTION public.enqueue_email(queue_name TEXT, payload JSONB)
|
||||
RETURNS BIGINT
|
||||
LANGUAGE plpgsql SECURITY DEFINER
|
||||
AS $$
|
||||
BEGIN
|
||||
RETURN pgmq.send(queue_name, payload);
|
||||
EXCEPTION WHEN undefined_table THEN
|
||||
PERFORM pgmq.create(queue_name);
|
||||
RETURN pgmq.send(queue_name, payload);
|
||||
END;
|
||||
$$;
|
||||
|
||||
CREATE OR REPLACE FUNCTION public.read_email_batch(queue_name TEXT, batch_size INT, vt INT)
|
||||
RETURNS TABLE(msg_id BIGINT, read_ct INT, message JSONB)
|
||||
LANGUAGE plpgsql SECURITY DEFINER
|
||||
AS $$
|
||||
BEGIN
|
||||
RETURN QUERY SELECT r.msg_id, r.read_ct, r.message FROM pgmq.read(queue_name, vt, batch_size) r;
|
||||
EXCEPTION WHEN undefined_table THEN
|
||||
PERFORM pgmq.create(queue_name);
|
||||
RETURN;
|
||||
END;
|
||||
$$;
|
||||
|
||||
CREATE OR REPLACE FUNCTION public.delete_email(queue_name TEXT, message_id BIGINT)
|
||||
RETURNS BOOLEAN
|
||||
LANGUAGE plpgsql SECURITY DEFINER
|
||||
AS $$
|
||||
BEGIN
|
||||
RETURN pgmq.delete(queue_name, message_id);
|
||||
EXCEPTION WHEN undefined_table THEN
|
||||
RETURN FALSE;
|
||||
END;
|
||||
$$;
|
||||
|
||||
CREATE OR REPLACE FUNCTION public.move_to_dlq(
|
||||
source_queue TEXT, dlq_name TEXT, message_id BIGINT, payload JSONB
|
||||
)
|
||||
RETURNS BIGINT
|
||||
LANGUAGE plpgsql SECURITY DEFINER
|
||||
AS $$
|
||||
DECLARE new_id BIGINT;
|
||||
BEGIN
|
||||
SELECT pgmq.send(dlq_name, payload) INTO new_id;
|
||||
PERFORM pgmq.delete(source_queue, message_id);
|
||||
RETURN new_id;
|
||||
EXCEPTION WHEN undefined_table THEN
|
||||
BEGIN
|
||||
PERFORM pgmq.create(dlq_name);
|
||||
EXCEPTION WHEN OTHERS THEN
|
||||
NULL;
|
||||
END;
|
||||
SELECT pgmq.send(dlq_name, payload) INTO new_id;
|
||||
BEGIN
|
||||
PERFORM pgmq.delete(source_queue, message_id);
|
||||
EXCEPTION WHEN undefined_table THEN
|
||||
NULL;
|
||||
END;
|
||||
RETURN new_id;
|
||||
END;
|
||||
$$;
|
||||
|
||||
-- Restrict queue RPC wrappers to service_role only (SECURITY DEFINER runs as owner,
|
||||
-- so without this any authenticated user could manipulate the email queues)
|
||||
REVOKE EXECUTE ON FUNCTION public.enqueue_email(TEXT, JSONB) FROM PUBLIC;
|
||||
GRANT EXECUTE ON FUNCTION public.enqueue_email(TEXT, JSONB) TO service_role;
|
||||
|
||||
REVOKE EXECUTE ON FUNCTION public.read_email_batch(TEXT, INT, INT) FROM PUBLIC;
|
||||
GRANT EXECUTE ON FUNCTION public.read_email_batch(TEXT, INT, INT) TO service_role;
|
||||
|
||||
REVOKE EXECUTE ON FUNCTION public.delete_email(TEXT, BIGINT) FROM PUBLIC;
|
||||
GRANT EXECUTE ON FUNCTION public.delete_email(TEXT, BIGINT) TO service_role;
|
||||
|
||||
REVOKE EXECUTE ON FUNCTION public.move_to_dlq(TEXT, TEXT, BIGINT, JSONB) FROM PUBLIC;
|
||||
GRANT EXECUTE ON FUNCTION public.move_to_dlq(TEXT, TEXT, BIGINT, JSONB) TO service_role;
|
||||
|
||||
-- Suppressed emails table (tracks unsubscribes, bounces, complaints)
|
||||
-- Append-only: no DELETE or UPDATE policies to prevent bypassing suppression.
|
||||
CREATE TABLE IF NOT EXISTS public.suppressed_emails (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email TEXT NOT NULL,
|
||||
reason TEXT NOT NULL CHECK (reason IN ('unsubscribe', 'bounce', 'complaint')),
|
||||
metadata JSONB,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
UNIQUE(email)
|
||||
);
|
||||
|
||||
ALTER TABLE public.suppressed_emails ENABLE ROW LEVEL SECURITY;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can read suppressed emails"
|
||||
ON public.suppressed_emails FOR SELECT
|
||||
USING (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can insert suppressed emails"
|
||||
ON public.suppressed_emails FOR INSERT
|
||||
WITH CHECK (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_suppressed_emails_email ON public.suppressed_emails(email);
|
||||
|
||||
-- Email unsubscribe tokens table (one token per email address for unsubscribe links)
|
||||
-- No DELETE policy to prevent removing tokens. UPDATE allowed only to mark tokens as used.
|
||||
CREATE TABLE IF NOT EXISTS public.email_unsubscribe_tokens (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
token TEXT NOT NULL UNIQUE,
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
used_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
ALTER TABLE public.email_unsubscribe_tokens ENABLE ROW LEVEL SECURITY;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can read tokens"
|
||||
ON public.email_unsubscribe_tokens FOR SELECT
|
||||
USING (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can insert tokens"
|
||||
ON public.email_unsubscribe_tokens FOR INSERT
|
||||
WITH CHECK (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
DO $$ BEGIN
|
||||
CREATE POLICY "Service role can mark tokens as used"
|
||||
ON public.email_unsubscribe_tokens FOR UPDATE
|
||||
USING (auth.role() = 'service_role')
|
||||
WITH CHECK (auth.role() = 'service_role');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_unsubscribe_tokens_token ON public.email_unsubscribe_tokens(token);
|
||||
|
||||
-- ============================================================
|
||||
-- POST-MIGRATION STEPS (applied dynamically by setup_email_infra)
|
||||
-- These steps contain project-specific secrets and URLs and
|
||||
-- cannot be expressed as static SQL. They are applied via the
|
||||
-- Supabase Management API (ExecuteSQL) each time the tool runs.
|
||||
-- ============================================================
|
||||
--
|
||||
-- 1. VAULT SECRET
|
||||
-- Stores (or updates) the Supabase service_role key in
|
||||
-- vault as 'email_queue_service_role_key'.
|
||||
-- Uses vault.create_secret / vault.update_secret (upsert).
|
||||
-- To revert: DELETE FROM vault.secrets WHERE name = 'email_queue_service_role_key';
|
||||
--
|
||||
-- 2. CRON JOB (pg_cron)
|
||||
-- Creates job 'process-email-queue' with a 5-second interval.
|
||||
-- The job checks:
|
||||
-- a) rate-limit cooldown (email_send_state.retry_after_until)
|
||||
-- b) whether auth_emails or transactional_emails queues have messages
|
||||
-- If conditions are met, it calls the process-email-queue Edge Function
|
||||
-- via net.http_post using the vault-stored service_role key.
|
||||
-- To revert: SELECT cron.unschedule('process-email-queue');
|
||||
Reference in New Issue
Block a user