diff --git a/frontend/app/auth/callback/route.ts b/frontend/app/auth/callback/route.ts index 8c48ac1c..0da7f129 100644 --- a/frontend/app/auth/callback/route.ts +++ b/frontend/app/auth/callback/route.ts @@ -1,7 +1,10 @@ import { NextRequest, NextResponse } from "next/server"; +import { BACKEND_ENTITLEMENT_HEADER } from "@/lib/backend-auth"; import { createSupabaseRouteClient, hasSupabaseServerEnv } from "@/lib/supabase/server"; import { getConfiguredSiteUrl } from "@/lib/site-url"; +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; + function normalizeNextPath(input: string | null) { const fallback = "/"; const raw = String(input || "").trim(); @@ -11,6 +14,34 @@ function normalizeNextPath(input: string | null) { return raw; } +async function warmSignupTrial(accessToken: string) { + const token = String(accessToken || "").trim(); + if (!API_BASE || !token) return; + + const headers = new Headers({ + Accept: "application/json", + Authorization: `Bearer ${token}`, + }); + const backendToken = process.env.POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN?.trim(); + if (backendToken) { + headers.set(BACKEND_ENTITLEMENT_HEADER, backendToken); + } + + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), 2500); + try { + await fetch(`${API_BASE}/api/auth/me`, { + cache: "no-store", + headers, + signal: controller.signal, + }); + } catch { + // The account/terminal bootstrap will retry. Callback must not strand login. + } finally { + clearTimeout(timeoutId); + } +} + export async function GET(request: NextRequest) { const siteUrl = getConfiguredSiteUrl(); if (siteUrl) { @@ -38,7 +69,10 @@ export async function GET(request: NextRequest) { const supabase = createSupabaseRouteClient(request, response); const code = request.nextUrl.searchParams.get("code"); if (code) { - await supabase.auth.exchangeCodeForSession(code); + const { + data: { session }, + } = await supabase.auth.exchangeCodeForSession(code); + await warmSignupTrial(session?.access_token || ""); } return response; diff --git a/frontend/components/auth/__tests__/passwordReset.test.ts b/frontend/components/auth/__tests__/passwordReset.test.ts index ff7d0c73..62ac5fad 100644 --- a/frontend/components/auth/__tests__/passwordReset.test.ts +++ b/frontend/components/auth/__tests__/passwordReset.test.ts @@ -26,6 +26,13 @@ export function runTests() { "auth", "ResetPasswordClient.tsx", ); + const authCallbackPath = path.join( + projectRoot, + "app", + "auth", + "callback", + "route.ts", + ); const loginClientSource = fs.readFileSync(loginClientPath, "utf8"); @@ -55,4 +62,14 @@ export function runTests() { resetClientSource.includes("expired"), "password reset client must detect an expired or invalid recovery session", ); + + const authCallbackSource = fs.readFileSync(authCallbackPath, "utf8"); + assert( + authCallbackSource.includes("warmSignupTrial") && + authCallbackSource.includes("exchangeCodeForSession") && + authCallbackSource.includes("/api/auth/me") && + authCallbackSource.includes("Authorization") && + authCallbackSource.includes("Bearer"), + "auth callback must warm /api/auth/me with the exchanged Supabase session so new users receive the signup trial immediately", + ); } diff --git a/scripts/supabase/schema.sql b/scripts/supabase/schema.sql index eb099789..58a927b2 100644 --- a/scripts/supabase/schema.sql +++ b/scripts/supabase/schema.sql @@ -46,6 +46,11 @@ create index if not exists idx_subscriptions_user_created on public.subscriptions(user_id, created_at desc) include (id, status, plan_code, source, starts_at, expires_at, updated_at); +create unique index if not exists uq_subscriptions_signup_trial_user + on public.subscriptions(user_id) + where plan_code = 'signup_trial_3d' + and source = 'signup_trial'; + create table if not exists public.payments ( id bigserial primary key, user_id uuid not null references auth.users(id) on delete cascade, @@ -128,6 +133,221 @@ create table if not exists public.trial_claim_wallets ( create unique index if not exists uq_trial_claim_wallets_address on public.trial_claim_wallets(lower(wallet_address)); +create or replace function public.claim_signup_trial( + p_user_id uuid, + p_email text default '', + p_telegram_user_id bigint default null, + p_wallet_addresses text[] default array[]::text[] +) +returns jsonb +language plpgsql +security definer +set search_path = public, auth +as $$ +declare + v_email text := lower(trim(coalesce(p_email, ''))); + v_wallets text[] := ( + select coalesce(array_agg(distinct lower(trim(input_wallet.value))), array[]::text[]) + from unnest(coalesce(p_wallet_addresses, array[]::text[])) as input_wallet(value) + where trim(input_wallet.value) <> '' + ); + v_now timestamptz := now(); + v_expires timestamptz := v_now + interval '3 days'; + v_claim public.trial_claims%rowtype; + v_claim_id bigint; + v_rows integer := 0; +begin + if p_user_id is null then + return jsonb_build_object('created', false, 'reason', 'missing_user_id'); + end if; + + select tc.* + into v_claim + from public.trial_claims tc + where tc.user_id = p_user_id + or (v_email <> '' and lower(tc.email) = v_email) + or (p_telegram_user_id is not null and tc.telegram_user_id = p_telegram_user_id) + or exists ( + select 1 + from public.trial_claim_wallets tcw + where tcw.trial_claim_id = tc.id + and lower(tcw.wallet_address) = any(v_wallets) + ) + order by + case when tc.user_id = p_user_id then 0 else 1 end, + tc.created_at asc, + tc.id asc + limit 1; + + if found then + if v_claim.user_id <> p_user_id then + return jsonb_build_object('created', false, 'reason', 'already_claimed'); + end if; + + insert into public.subscriptions ( + user_id, + plan_code, + status, + starts_at, + expires_at, + source, + created_at, + updated_at + ) + values ( + p_user_id, + 'signup_trial_3d', + 'active', + coalesce(v_claim.claimed_at, v_now), + coalesce(v_claim.claimed_at, v_now) + interval '3 days', + 'signup_trial', + v_now, + v_now + ) + on conflict (user_id) + where plan_code = 'signup_trial_3d' + and source = 'signup_trial' + do nothing; + get diagnostics v_rows = row_count; + + if v_rows > 0 then + insert into public.entitlement_events ( + user_id, + action, + reason, + actor, + payload, + created_at + ) + values ( + p_user_id, + 'signup_trial_granted', + 'claim_repaired', + 'supabase_auth', + jsonb_build_object( + 'plan_code', 'signup_trial_3d', + 'expires_at', coalesce(v_claim.claimed_at, v_now) + interval '3 days' + ), + v_now + ); + return jsonb_build_object( + 'created', true, + 'repaired', true, + 'plan_code', 'signup_trial_3d', + 'expires_at', coalesce(v_claim.claimed_at, v_now) + interval '3 days' + ); + end if; + + return jsonb_build_object('created', false, 'reason', 'already_claimed'); + end if; + + insert into public.trial_claims ( + user_id, + email, + telegram_user_id, + primary_wallet_address, + metadata, + claimed_at, + created_at + ) + values ( + p_user_id, + v_email, + p_telegram_user_id, + nullif(v_wallets[1], ''), + jsonb_build_object('wallet_addresses', v_wallets), + v_now, + v_now + ) + returning id into v_claim_id; + + insert into public.trial_claim_wallets (trial_claim_id, wallet_address, created_at) + select v_claim_id, wallet.value, v_now + from unnest(v_wallets) as wallet(value) + on conflict do nothing; + + insert into public.subscriptions ( + user_id, + plan_code, + status, + starts_at, + expires_at, + source, + created_at, + updated_at + ) + values ( + p_user_id, + 'signup_trial_3d', + 'active', + v_now, + v_expires, + 'signup_trial', + v_now, + v_now + ) + on conflict (user_id) + where plan_code = 'signup_trial_3d' + and source = 'signup_trial' + do nothing; + get diagnostics v_rows = row_count; + + insert into public.entitlement_events ( + user_id, + action, + reason, + actor, + payload, + created_at + ) + values + ( + p_user_id, + 'signup_trial_claimed', + 'trial_dedupe', + 'supabase_auth', + jsonb_build_object( + 'user_id', p_user_id, + 'email', v_email, + 'telegram_user_id', p_telegram_user_id, + 'wallet_addresses', v_wallets, + 'claimed_at', v_now, + 'storage', 'trial_claims' + ), + v_now + ), + ( + p_user_id, + 'signup_trial_granted', + 'first_auth', + 'supabase_auth', + jsonb_build_object( + 'plan_code', 'signup_trial_3d', + 'expires_at', v_expires + ), + v_now + ); + + if v_rows = 0 then + return jsonb_build_object('created', false, 'reason', 'already_claimed'); + end if; + + return jsonb_build_object( + 'created', true, + 'plan_code', 'signup_trial_3d', + 'expires_at', v_expires + ); +exception + when unique_violation then + return jsonb_build_object('created', false, 'reason', 'already_claimed'); +end; +$$; + +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from public; +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from anon; +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from authenticated; +grant execute on function public.claim_signup_trial(uuid, text, bigint, text[]) to service_role; + create table if not exists public.referral_codes ( id bigserial primary key, user_id uuid not null references auth.users(id) on delete cascade, diff --git a/scripts/supabase/signup_trial_atomic_20260530.sql b/scripts/supabase/signup_trial_atomic_20260530.sql new file mode 100644 index 00000000..b5be0d15 --- /dev/null +++ b/scripts/supabase/signup_trial_atomic_20260530.sql @@ -0,0 +1,248 @@ +-- Atomic signup-trial claim and subscription grant. +-- Run after subscription_referral_20260529.sql. + +-- Keep the first real signup trial row per user and neutralize duplicate rows +-- before adding the one-trial-per-user index. +with ranked_signup_trials as ( + select + id, + row_number() over ( + partition by user_id + order by created_at asc, id asc + ) as rn + from public.subscriptions + where plan_code = 'signup_trial_3d' + and source = 'signup_trial' +) +update public.subscriptions s +set + status = case when s.status = 'active' then 'expired' else s.status end, + source = 'signup_trial_duplicate', + updated_at = now() +from ranked_signup_trials r +where s.id = r.id + and r.rn > 1; + +create unique index if not exists uq_subscriptions_signup_trial_user + on public.subscriptions(user_id) + where plan_code = 'signup_trial_3d' + and source = 'signup_trial'; + +create or replace function public.claim_signup_trial( + p_user_id uuid, + p_email text default '', + p_telegram_user_id bigint default null, + p_wallet_addresses text[] default array[]::text[] +) +returns jsonb +language plpgsql +security definer +set search_path = public, auth +as $$ +declare + v_email text := lower(trim(coalesce(p_email, ''))); + v_wallets text[] := ( + select coalesce(array_agg(distinct lower(trim(input_wallet.value))), array[]::text[]) + from unnest(coalesce(p_wallet_addresses, array[]::text[])) as input_wallet(value) + where trim(input_wallet.value) <> '' + ); + v_now timestamptz := now(); + v_expires timestamptz := v_now + interval '3 days'; + v_claim public.trial_claims%rowtype; + v_claim_id bigint; + v_rows integer := 0; +begin + if p_user_id is null then + return jsonb_build_object('created', false, 'reason', 'missing_user_id'); + end if; + + select tc.* + into v_claim + from public.trial_claims tc + where tc.user_id = p_user_id + or (v_email <> '' and lower(tc.email) = v_email) + or (p_telegram_user_id is not null and tc.telegram_user_id = p_telegram_user_id) + or exists ( + select 1 + from public.trial_claim_wallets tcw + where tcw.trial_claim_id = tc.id + and lower(tcw.wallet_address) = any(v_wallets) + ) + order by + case when tc.user_id = p_user_id then 0 else 1 end, + tc.created_at asc, + tc.id asc + limit 1; + + if found then + if v_claim.user_id <> p_user_id then + return jsonb_build_object('created', false, 'reason', 'already_claimed'); + end if; + + insert into public.subscriptions ( + user_id, + plan_code, + status, + starts_at, + expires_at, + source, + created_at, + updated_at + ) + values ( + p_user_id, + 'signup_trial_3d', + 'active', + coalesce(v_claim.claimed_at, v_now), + coalesce(v_claim.claimed_at, v_now) + interval '3 days', + 'signup_trial', + v_now, + v_now + ) + on conflict (user_id) + where plan_code = 'signup_trial_3d' + and source = 'signup_trial' + do nothing; + get diagnostics v_rows = row_count; + + if v_rows > 0 then + insert into public.entitlement_events ( + user_id, + action, + reason, + actor, + payload, + created_at + ) + values ( + p_user_id, + 'signup_trial_granted', + 'claim_repaired', + 'supabase_auth', + jsonb_build_object( + 'plan_code', 'signup_trial_3d', + 'expires_at', coalesce(v_claim.claimed_at, v_now) + interval '3 days' + ), + v_now + ); + return jsonb_build_object( + 'created', true, + 'repaired', true, + 'plan_code', 'signup_trial_3d', + 'expires_at', coalesce(v_claim.claimed_at, v_now) + interval '3 days' + ); + end if; + + return jsonb_build_object('created', false, 'reason', 'already_claimed'); + end if; + + insert into public.trial_claims ( + user_id, + email, + telegram_user_id, + primary_wallet_address, + metadata, + claimed_at, + created_at + ) + values ( + p_user_id, + v_email, + p_telegram_user_id, + nullif(v_wallets[1], ''), + jsonb_build_object('wallet_addresses', v_wallets), + v_now, + v_now + ) + returning id into v_claim_id; + + insert into public.trial_claim_wallets (trial_claim_id, wallet_address, created_at) + select v_claim_id, wallet.value, v_now + from unnest(v_wallets) as wallet(value) + on conflict do nothing; + + insert into public.subscriptions ( + user_id, + plan_code, + status, + starts_at, + expires_at, + source, + created_at, + updated_at + ) + values ( + p_user_id, + 'signup_trial_3d', + 'active', + v_now, + v_expires, + 'signup_trial', + v_now, + v_now + ) + on conflict (user_id) + where plan_code = 'signup_trial_3d' + and source = 'signup_trial' + do nothing; + get diagnostics v_rows = row_count; + + insert into public.entitlement_events ( + user_id, + action, + reason, + actor, + payload, + created_at + ) + values + ( + p_user_id, + 'signup_trial_claimed', + 'trial_dedupe', + 'supabase_auth', + jsonb_build_object( + 'user_id', p_user_id, + 'email', v_email, + 'telegram_user_id', p_telegram_user_id, + 'wallet_addresses', v_wallets, + 'claimed_at', v_now, + 'storage', 'trial_claims' + ), + v_now + ), + ( + p_user_id, + 'signup_trial_granted', + 'first_auth', + 'supabase_auth', + jsonb_build_object( + 'plan_code', 'signup_trial_3d', + 'expires_at', v_expires + ), + v_now + ); + + if v_rows = 0 then + return jsonb_build_object('created', false, 'reason', 'already_claimed'); + end if; + + return jsonb_build_object( + 'created', true, + 'plan_code', 'signup_trial_3d', + 'expires_at', v_expires + ); +exception + when unique_violation then + return jsonb_build_object('created', false, 'reason', 'already_claimed'); +end; +$$; + +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from public; +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from anon; +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from authenticated; +grant execute on function public.claim_signup_trial(uuid, text, bigint, text[]) to service_role; + +analyze public.trial_claims; +analyze public.trial_claim_wallets; +analyze public.subscriptions; diff --git a/scripts/supabase/signup_trial_atomic_lockdown_20260530.sql b/scripts/supabase/signup_trial_atomic_lockdown_20260530.sql new file mode 100644 index 00000000..1d9830ba --- /dev/null +++ b/scripts/supabase/signup_trial_atomic_lockdown_20260530.sql @@ -0,0 +1,11 @@ +-- Restrict atomic signup-trial RPC to backend service-role calls only. + +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from public; +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from anon; +revoke all on function public.claim_signup_trial(uuid, text, bigint, text[]) from authenticated; +grant execute on function public.claim_signup_trial(uuid, text, bigint, text[]) to service_role; + +select + has_function_privilege('anon', 'public.claim_signup_trial(uuid,text,bigint,text[])', 'execute') as anon_can_execute, + has_function_privilege('authenticated', 'public.claim_signup_trial(uuid,text,bigint,text[])', 'execute') as authenticated_can_execute, + has_function_privilege('service_role', 'public.claim_signup_trial(uuid,text,bigint,text[])', 'execute') as service_role_can_execute; diff --git a/scripts/supabase/signup_trial_atomic_verify.sql b/scripts/supabase/signup_trial_atomic_verify.sql new file mode 100644 index 00000000..cd51928d --- /dev/null +++ b/scripts/supabase/signup_trial_atomic_verify.sql @@ -0,0 +1,38 @@ +-- Read-only verification for signup_trial_atomic_20260530.sql. + +select + exists ( + select 1 + from pg_indexes + where schemaname = 'public' + and tablename = 'subscriptions' + and indexname = 'uq_subscriptions_signup_trial_user' + ) as has_trial_unique_index, + exists ( + select 1 + from pg_proc p + join pg_namespace n on n.oid = p.pronamespace + where n.nspname = 'public' + and p.proname = 'claim_signup_trial' + ) as has_claim_signup_trial_rpc; + +select + count(*) as users_with_duplicate_signup_trials, + coalesce(sum(trial_count), 0) as total_trial_rows_in_duplicate_users +from ( + select user_id, count(*) as trial_count + from public.subscriptions + where plan_code = 'signup_trial_3d' + and source = 'signup_trial' + group by user_id + having count(*) > 1 +) d; + +select + count(*) filter (where s.id is null) as trial_claims_without_trial_subscription, + count(*) as total_trial_claims +from public.trial_claims tc +left join public.subscriptions s + on s.user_id = tc.user_id + and s.plan_code = 'signup_trial_3d' + and s.source = 'signup_trial'; diff --git a/src/auth/supabase_entitlement.py b/src/auth/supabase_entitlement.py index 0de4621a..465f1a65 100644 --- a/src/auth/supabase_entitlement.py +++ b/src/auth/supabase_entitlement.py @@ -176,6 +176,31 @@ class SupabaseEntitlementService: except Exception: return None + def _rpc( + self, + name: str, + payload: Optional[Any] = None, + *, + allowed_status: Optional[List[int]] = None, + ) -> Any: + return self._rest( + "POST", + f"rpc/{name}", + payload=payload or {}, + allowed_status=allowed_status or [200], + ) + + @staticmethod + def _looks_like_missing_rpc(exc: Exception) -> bool: + text = str(exc).lower() + return ( + "pgrst202" in text + or "could not find the function" in text + or "function public.claim_signup_trial" in text + or "schema cache" in text + or "404" in text + ) + def _admin_user_endpoint(self, user_id: str) -> str: return f"{self.supabase_url}/auth/v1/admin/users/{user_id}" @@ -407,6 +432,31 @@ class SupabaseEntitlementService: try: telegram_user_id = self._telegram_user_id_for(user_key) wallet_addresses = self._active_wallet_addresses_for(user_key) + try: + result = self._rpc( + "claim_signup_trial", + { + "p_user_id": user_key, + "p_email": normalized_email, + "p_telegram_user_id": telegram_user_id, + "p_wallet_addresses": wallet_addresses, + }, + allowed_status=[200], + ) + if isinstance(result, dict): + self.invalidate_subscription_cache(user_key) + return result + if isinstance(result, list) and result and isinstance(result[0], dict): + self.invalidate_subscription_cache(user_key) + return result[0] + except Exception as rpc_exc: + if not self._looks_like_missing_rpc(rpc_exc): + raise + logger.warning( + "signup trial rpc missing; falling back to legacy grant user_id={}: {}", + user_key, + rpc_exc, + ) if self._trial_claim_exists( user_id=user_key, email=normalized_email, @@ -437,6 +487,13 @@ class SupabaseEntitlementService: if isinstance(claim_rows, list) and claim_rows and isinstance(claim_rows[0], dict): claim_id = claim_rows[0].get("id") except Exception: + if self._trial_claim_exists( + user_id=user_key, + email=normalized_email, + telegram_user_id=telegram_user_id, + wallet_addresses=wallet_addresses, + ): + return {"created": False, "reason": "already_claimed"} self._record_signup_trial_claim_event( user_id=user_key, email=normalized_email, diff --git a/tests/test_subscription_referral_program.py b/tests/test_subscription_referral_program.py index 17b753fc..f4b68a46 100644 --- a/tests/test_subscription_referral_program.py +++ b/tests/test_subscription_referral_program.py @@ -78,6 +78,14 @@ def test_signup_trial_claim_creates_three_day_trial_once(monkeypatch): service = SupabaseEntitlementService() calls = [] + monkeypatch.setattr( + service, + "_rpc", + lambda *args, **kwargs: (_ for _ in ()).throw( + RuntimeError("PGRST202 function not found") + ), + ) + def fake_rest(method, table, **kwargs): calls.append({"method": method, "table": table, **kwargs}) if method == "GET" and table in {"trial_claims", "user_wallets"}: @@ -99,6 +107,49 @@ def test_signup_trial_claim_creates_three_day_trial_once(monkeypatch): assert payload["user_id"] == "user-1" +def test_signup_trial_uses_atomic_database_claim_rpc(monkeypatch): + monkeypatch.setenv("POLYWEATHER_AUTH_ENABLED", "true") + monkeypatch.setenv("SUPABASE_URL", "https://example.supabase.co") + monkeypatch.setenv("SUPABASE_ANON_KEY", "anon") + monkeypatch.setenv("SUPABASE_SERVICE_ROLE_KEY", "service-role") + service = SupabaseEntitlementService() + rpc_calls = [] + + def fake_rpc(name, payload, **kwargs): + rpc_calls.append({"name": name, "payload": payload, **kwargs}) + return { + "created": True, + "plan_code": "signup_trial_3d", + "expires_at": "2026-06-02T00:00:00+00:00", + } + + def fake_rest(method, table, **kwargs): + if method == "GET" and table == "user_wallets": + return [{"address": "0xabc"}] + raise AssertionError( + f"signup trial must use atomic RPC instead of legacy {method} {table}" + ) + + monkeypatch.setattr(service, "_rpc", fake_rpc, raising=False) + monkeypatch.setattr(service, "_rest", fake_rest) + + result = service.ensure_signup_trial("user-1", "USER@example.com") + + assert result["created"] is True + assert rpc_calls == [ + { + "name": "claim_signup_trial", + "payload": { + "p_user_id": "user-1", + "p_email": "user@example.com", + "p_telegram_user_id": None, + "p_wallet_addresses": ["0xabc"], + }, + "allowed_status": [200], + } + ] + + def test_referral_discount_applies_to_first_monthly_payment(monkeypatch, tmp_path): _payment_env(monkeypatch, tmp_path) service = PaymentContractCheckoutService() @@ -187,6 +238,14 @@ def test_signup_trial_falls_back_to_entitlement_events_without_trial_tables(monk service = SupabaseEntitlementService() calls = [] + monkeypatch.setattr( + service, + "_rpc", + lambda *args, **kwargs: (_ for _ in ()).throw( + RuntimeError("PGRST202 function not found") + ), + ) + def fake_rest(method, table, **kwargs): calls.append({"method": method, "table": table, **kwargs}) if method == "GET" and table == "user_wallets":