admin/mylegal-stage-law
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Changes

Browse Source 
Co-authored-by: renee-png <262607627+renee-png@users.noreply.github.com>
This commit is contained in:
gpt-engineer-app[bot]
2026-04-23 20:55:30 +00:00
co-authored by renee-png
parent 24850e3bf5
commit 98442e5982
2 changed files with 618 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/lovable/email/queue/process.ts
+326
View File
@@ -0,0 +1,326 @@
import { sendLovableEmail } from '@lovable.dev/email-js'
import { createClient } from '@supabase/supabase-js'
import { createFileRoute } from '@tanstack/react-router'
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: ReturnType<typeof createClient>,
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(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 })
},
},
},
})
supabase/migrations/20260423205526_email_infra.sql
+292
View File
@@ -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');