diff --git a/bun.lockb b/bun.lockb index 328f248..883ed2f 100755 Binary files a/bun.lockb and b/bun.lockb differ diff --git a/package.json b/package.json index 194e874..b1a82ca 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/src/routeTree.gen.ts b/src/routeTree.gen.ts index 1983a7d..3caf79d 100644 --- a/src/routeTree.gen.ts +++ b/src/routeTree.gen.ts @@ -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() - -import type { getRouter } from './router.tsx' -import type { createStart } from '@tanstack/react-start' -declare module '@tanstack/react-start' { - interface Register { - ssr: true - router: Awaited> - } -} diff --git a/src/routes/lovable/email/queue/process.ts b/src/routes/lovable/email/queue/process.ts new file mode 100644 index 0000000..c424679 --- /dev/null +++ b/src/routes/lovable/email/queue/process.ts @@ -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> + +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 }, + reason: string +): Promise { + 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(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 = { + 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() + 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 }) + }, + }, + }, +}) diff --git a/supabase/migrations/20260423205526_email_infra.sql b/supabase/migrations/20260423205526_email_infra.sql new file mode 100644 index 0000000..e0f967a --- /dev/null +++ b/supabase/migrations/20260423205526_email_infra.sql @@ -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');