Expo + Supabase 冷启动安全架构

A) Architecture summary - **Expo client**: Handles UI, optimistic updates, and real-time subscription to job status via Supabase Realtime. It never blocks on network calls; all Edge Function calls are fire-and-forget or non-blocking. - **Supabase Postgres**: Stores `jobs`, `generations`, and `entitlements` tables with RLS for user isolation. Indexes on status and user_id for efficient polling. - **Supabase Edge Functions**: Lightweight, stateless functions for auth validation, credit checks, job enqueue, and status reads. No heavy SDKs; use raw fetch and Supabase client with service role key (or user JWT). - **Worker service**: A separate Node.js/Deno process (or cron) that polls `jobs` table for `queued` status, runs AI generation (e.g., via OpenAI API), uploads to Supabase Storage, and updates job + generation records. Handles retries and idempotency via `attempt_count` and `status` transitions. B) Migrations (SQL) ```sql -- 1. Create extensions create extension if not exists "pgcrypto"; -- 2. Entitlements table (credits and paid status) create table public.entitlements ( id uuid primary key default gen_random_uuid(), user_id uuid not null references auth.users(id) on delete cascade, credits_remaining integer not null default 0, is_paid boolean not null default false, updated_at timestamptz not null default now(), unique(user_id) ); -- 3. Jobs table (queue for worker) create table public.jobs ( id uuid primary key default gen_random_uuid(), user_id uuid not null references auth.users(id) on delete cascade, status text not null default 'queued' check (status in ('queued', 'processing', 'completed', 'failed')), input_prompt text not null, attempt_count integer not null default 0, max_attempts integer not null default 3, error_message text, created_at timestamptz not null default now(), updated_at timestamptz not null default now() ); -- 4. Generations table (final output) create table public.generations ( id uuid primary key default gen_random_uuid(), job_id uuid not null references public.jobs(id) on delete cascade, user_id uuid not null references auth.users(id) on delete cascade, storage_path text, original_input text not null, status text not null default 'pending' check (status in ('pending', 'completed', 'failed')), created_at timestamptz not null default now() ); -- 5. Indexes for performance create index idx_jobs_status on public.jobs(status) where status = 'queued'; create index idx_jobs_user_id on public.jobs(user_id); create index idx_generations_user_id on public.generations(user_id); create index idx_generations_job_id on public.generations(job_id); -- 6. RLS policies (enable on all tables) alter table public.entitlements enable row level security; alter table public.jobs enable row level security; alter table public.generations enable row level security; -- Entitlements: user can read/update own create policy "entitlements_select_own" on public.entitlements for select using (auth.uid() = user_id); create policy "entitlements_update_own" on public.entitlements for update using (auth.uid() = user_id); -- Jobs: user can read own, insert own, update own (for status) create policy "jobs_select_own" on public.jobs for select using (auth.uid() = user_id); create policy "jobs_insert_own" on public.jobs for insert with check (auth.uid() = user_id); create policy "jobs_update_own" on public.jobs for update using (auth.uid() = user_id); -- Generations: user can read own, insert own (worker uses service role) create policy "generations_select_own" on public.generations for select using (auth.uid() = user_id); create policy "generations_insert_own" on public.generations for insert with check (auth.uid() = user_id); ``` C) Edge function file structure + key code blocks ``` supabase/functions/ ├── ping/index.ts ├── enqueue_generation/index.ts └── get_job_status/index.ts ``` **ping/index.ts** (no auth, just health check) ```typescript import { serve } from "https://deno.land/std@0.177.0/http/server.ts"; serve(async (req) => { if (req.method === "HEAD" || req.method === "GET") { return new Response("ok", { status: 200 }); } return new Response("Method not allowed", { status: 405 }); }); ``` **enqueue_generation/index.ts** (validate auth, check credits, create job) ```typescript import { serve } from "https://deno.land/std@0.177.0/http/server.ts"; import { createClient } from "https://esm.sh/@supabase/supabase-js@2"; const supabase = createClient( Deno.env.get("SUPABASE_URL")!, Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")! ); serve(async (req) => { if (req.method !== "POST") return new Response("Method not allowed", { status: 405 }); const authHeader = req.headers.get("Authorization"); if (!authHeader) return new Response("Unauthorized", { status: 401 }); const token = authHeader.replace("Bearer ", ""); const { data: { user }, error: authError } = await supabase.auth.getUser(token); if (authError || !user) return new Response("Unauthorized", { status: 401 }); const { prompt } = await req.json(); if (!prompt || prompt.length > 500) return new Response("Invalid prompt", { status: 400 }); // Check entitlements const { data: entitlement, error: entError } = await supabase .from("entitlements") .select("credits_remaining, is_paid") .eq("user_id", user.id) .single(); if (entError || !entitlement || (!entitlement.is_paid && entitlement.credits_remaining <= 0)) { return new Response("Insufficient credits", { status: 403 }); } // Create job const { data: job, error: jobError } = await supabase .from("jobs") .insert({ user_id: user.id, input_prompt: prompt }) .select() .single(); if (jobError) return new Response("Failed to create job", { status: 500 }); // Decrement credits (non-blocking, best-effort) await supabase .from("entitlements") .update({ credits_remaining: entitlement.credits_remaining - 1 }) .eq("user_id", user.id); return new Response(JSON.stringify({ jobId: job.id }), { status: 200, headers: { "Content-Type": "application/json" } }); }); ``` **get_job_status/index.ts** (light read) ```typescript import { serve } from "https://deno.land/std@0.177.0/http/server.ts"; import { createClient } from "https://esm.sh/@supabase/supabase-js@2"; const supabase = createClient( Deno.env.get("SUPABASE_URL")!, Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")! ); serve(async (req) => { const url = new URL(req.url); const jobId = url.searchParams.get("jobId"); if (!jobId) return new Response("Missing jobId", { status: 400 }); const authHeader = req.headers.get("Authorization"); if (!authHeader) return new Response("Unauthorized", { status: 401 }); const token = authHeader.replace("Bearer ", ""); const { data: { user }, error: authError } = await supabase.auth.getUser(token); if (authError || !user) return new Response("Unauthorized", { status: 401 }); const { data: job, error } = await supabase .from("jobs") .select("status, error_message, generations(storage_path, original_input)") .eq("id", jobId) .eq("user_id", user.id) .single(); if (error) return new Response("Job not found", { status: 404 }); return new Response(JSON.stringify(job), { status: 200, headers: { "Content-Type": "application/json" } }); }); ``` D) Expo integration notes + key code blocks - **Warm ping on app start**: Use `useEffect` with `fetch` to call the ping function, but never `await` it in the render path. Wrap in a try/catch and ignore errors. - **Generate button**: Optimistic UI – immediately add a placeholder generation to local state, then call `enqueue_generation` in the background. - **Realtime subscription**: Subscribe to `jobs` table changes for the current user via Supabase Realtime. Fallback to polling every 5 seconds if Realtime fails. - **Gallery update**: When job status becomes `completed`, replace the placeholder with the actual generation from the `generations` table. ```typescript // app/_layout.tsx (or root component) import { useEffect } from "react"; import { supabase } from "@/lib/supabase"; export default function RootLayout() { useEffect(() => { // Non-blocking warm ping fetch(`${process.env.EXPO_PUBLIC_SUPABASE_URL}/functions/v1/ping`) .catch(() => {}); // ignore errors }, []); return ; } // components/GenerateButton.tsx import { useState } from "react"; import { supabase } from "@/lib/supabase"; export function GenerateButton({ prompt }: { prompt: string }) { const [isGenerating, setIsGenerating] = useState(false); const handleGenerate = async () => { setIsGenerating(true); // Optimistic UI: add placeholder const placeholderId = `temp-${Date.now()}`; addPlaceholderToGallery(placeholderId, prompt); try { const { data: { session } } = await supabase.auth.getSession(); const res = await fetch(`${process.env.EXPO_PUBLIC_SUPABASE_URL}/functions/v1/enqueue_generation`, { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${session?.access_token}`, }, body: JSON.stringify({ prompt }), }); if (!res.ok) throw new Error("Failed to enqueue"); const { jobId } = await res.json(); // Subscribe to realtime updates for this job const channel = supabase .channel(`job-${jobId}`) .on("postgres_changes", { event: "UPDATE", schema: "public", table: "jobs", filter: `id=eq.${jobId}` }, (payload) => { if (payload.new.status === "completed") { // Fetch generation and replace placeholder fetchGeneration(jobId, placeholderId); channel.unsubscribe(); } else if (payload.new.status === "failed") { // Show error in placeholder markPlaceholderFailed(placeholderId, payload.new.error_message); channel.unsubscribe(); } }) .subscribe(); // Fallback polling (if realtime fails) setTimeout(() => pollJobStatus(jobId, placeholderId), 5000); } catch (error) { markPlaceholderFailed(placeholderId, "Network error"); } finally { setIsGenerating(false); } }; return ; } ``` E) Worker outline + pseudo-code **Worker service** (Node.js/Deno, runs as a separate process or cron) ``` worker/ ├── index.ts (main loop) ├── ai.ts (AI generation logic) └── storage.ts (Supabase storage upload) ``` **Pseudo-code** ```typescript // index.ts import { createClient } from "@supabase/supabase-js"; import { generateImage } from "./ai"; import { uploadToStorage } from "./storage"; const supabase = createClient(SUPABASE_URL, SERVICE_ROLE_KEY); async function processJobs() { // Fetch queued jobs (limit 5 per batch) const { data: jobs } = await supabase .from("jobs") .select("*") .eq("status", "queued") .limit(5); for (const job of jobs) { // Idempotency: mark as processing with attempt count const { error: updateError } = await supabase .from("jobs") .update({ status: "processing", attempt_count: job.attempt_count + 1 }) .eq("id", job.id) .eq("status", "queued"); // conditional update prevents double-processing if (updateError) continue; // another worker got it try { // Run AI generation const imageBuffer = await generateImage(job.input_prompt); // Upload to storage const storagePath = `generations/${job.user_id}/${job.id}.png`; await uploadToStorage(storagePath, imageBuffer); // Update job + insert generation await supabase.from("jobs").update({ status: "completed" }).eq("id", job.id); await supabase.from("generations").insert({ job_id: job.id, user_id: job.user_id, storage_path: storagePath, original_input: job.input_prompt, status: "completed", }); } catch (error) { // Retry policy if (job.attempt_count >= job.max_attempts) { await supabase.from("jobs").update({ status: "failed", error_message: error.message }).eq("id", job.id); // Always create a generation record with original input await supabase.from("generations").insert({ job_id: job.id, user_id: job.user_id, original_input: job.input_prompt, status: "failed", }); } else { // Re-queue for retry await supabase.from("jobs").update({ status: "queued" }).eq("id", job.id); } } } } // Main loop: run every 2 seconds setInterval(processJobs, 2000); ``` **Key design decisions**: - Conditional update (`eq("status", "queued")`) ensures only one worker picks up a job (idempotency). - Failed jobs always create a `generations` record with `original_input` so the user sees what was attempted. - Worker never blocks Edge Functions; it polls the DB directly. - No AI calls in Edge Functions – they only enqueue and read status.

Expo + Supabase 冷启动安全架构

提示词内容收藏

用法

参考资料

讨论