fix(billing): harden concurrency safety and payment failure handling

dashboard/lib/supabase/supabase.sql

@@ -7,6 +7,7 @@ CREATE TABLE public.organization_billing (
   stripe_subscription_id text UNIQUE,
   plan text NOT NULL DEFAULT 'free'::text,
   pending_plan text,  -- Scheduled plan change (e.g., downgrade at period end)
+  payment_status text NOT NULL DEFAULT 'ok'::text,  -- 'ok' | 'past_due' | 'blocked'
   period_end timestamp with time zone,
   created_at timestamp with time zone DEFAULT now(),
   updated_at timestamp with time zone DEFAULT now(),
@@ -115,6 +116,66 @@ CREATE TABLE public.usage_sync_logs (
   CONSTRAINT usage_sync_logs_project_id_fkey FOREIGN KEY (project_id) REFERENCES public.organization_projects(project_id)
 );
 
+-- Atomic increment for project_usage fields (avoids read-modify-write race conditions)
+-- Called from sync-usage Edge Function to safely increment usage counters.
+create or replace function public.increment_project_usage(
+  p_project_id uuid,
+  p_field text,
+  p_increment integer
+)
+returns void
+language plpgsql
+security definer
+set search_path = ''
+as $$
+begin
+  -- Allowlist: only known usage columns may be incremented.
+  if p_field not in ('current_task', 'current_skill', 'current_fast_skill_search', 'current_agentic_skill_search', 'current_storage') then
+    raise exception 'Invalid field name: %', p_field;
+  end if;
+
+  -- Single atomic upsert: INSERT or UPDATE in one statement.
+  -- Avoids race condition where two concurrent UPDATEs both find NOT FOUND
+  -- and then both attempt INSERT, causing a unique constraint violation.
+  execute format(
+    'INSERT INTO public.project_usage (project_id, %I) VALUES ($1, $2)
+     ON CONFLICT (project_id) DO UPDATE SET %I = COALESCE(public.project_usage.%I, 0) + EXCLUDED.%I,
+     updated_at = now()',
+    p_field, p_field, p_field, p_field
+  ) using p_project_id, p_increment;
+end;
+$$;
+
+-- Atomically claim a monthly reset checkpoint for an org.
+-- Returns true if this call claimed it (i.e., the timestamp was actually changed),
+-- false if it was already set to the current month's timestamp.
+-- Uses ON CONFLICT with a WHERE clause so the UPDATE only fires when the value differs.
+create or replace function public.try_claim_monthly_reset(
+  p_checkpoint_id text,
+  p_month_start_timestamp bigint
+)
+returns boolean
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_row_count integer;
+begin
+  INSERT INTO public.usage_sync_global_checkpoint (id, last_processed_to_timestamp)
+  VALUES (p_checkpoint_id, p_month_start_timestamp)
+  ON CONFLICT (id) DO UPDATE
+    SET last_processed_to_timestamp = EXCLUDED.last_processed_to_timestamp
+    WHERE public.usage_sync_global_checkpoint.last_processed_to_timestamp != EXCLUDED.last_processed_to_timestamp;
+
+  -- ROW_COUNT is 0 when the WHERE clause suppresses the update (already claimed),
+  -- 1 when a row was actually inserted or updated.
+  -- FOUND cannot be used here because it is true even when the conditional UPDATE is skipped.
+  GET DIAGNOSTICS v_row_count = ROW_COUNT;
+  return v_row_count > 0;
+end;
+$$;
+
 -- Database function to create organization
 -- Uses auth.uid() to get the current authenticated user
 create or replace function public.create_organization(

dashboard/supabase/functions/stripe-webhook/index.ts

@@ -167,6 +167,7 @@ Deno.serve(async (req: Request) => {
               stripe_subscription_id: null,
               period_end: null,
               pending_plan: null,
+              payment_status: "ok",
             })
             .eq("organization_id", organizationId);
 
@@ -198,31 +199,40 @@ Deno.serve(async (req: Request) => {
         // Check if subscription is scheduled for cancellation
         if (subscription.cancel_at_period_end) {
           const downgradeTarget = subscription.metadata.downgrade_to;
+          const validDowngradePlans = ["free", "pro", "team"];
+          const validatedDowngrade = (downgradeTarget && validDowngradePlans.includes(downgradeTarget)) ? downgradeTarget : undefined;
           console.log(
             `Subscription ${subscription.id} is scheduled for cancellation${
-              downgradeTarget ? ` (downgrade to ${downgradeTarget})` : ""
+              validatedDowngrade ? ` (downgrade to ${validatedDowngrade})` : ""
             }`
           );
 
-          // Update with pending_plan if downgrading
-          if (downgradeTarget) {
-            const { error } = await supabase
-              .from("organization_billing")
-              .update({
-                pending_plan: downgradeTarget,
-              })
-              .eq("organization_id", organizationId);
+          // Always update period_end and stripe_subscription_id even when scheduled for cancellation
+          const periodEnd = await getPeriodEnd(subscription);
+          const updateData: Record<string, unknown> = {
+            stripe_subscription_id: subscription.id,
+            period_end: periodEnd,
+          };
+          if (validatedDowngrade) {
+            updateData.pending_plan = validatedDowngrade;
+          }
 
-            if (error) {
-              console.error(
-                `Error setting pending_plan for organization ${organizationId}:`,
-                error
-              );
-            } else {
-              console.log(
-                `Set pending_plan to ${downgradeTarget} for organization ${organizationId}`
-              );
-            }
+          const { error } = await supabase
+            .from("organization_billing")
+            .update(updateData)
+            .eq("organization_id", organizationId);
+
+          if (error) {
+            console.error(
+              `Error updating organization ${organizationId} for scheduled cancellation:`,
+              error
+            );
+          } else {
+            console.log(
+              `Updated organization ${organizationId}: period_end=${periodEnd}${
+                validatedDowngrade ? `, pending_plan=${validatedDowngrade}` : ""
+              }`
+            );
           }
           break;
         }
@@ -245,6 +255,7 @@ Deno.serve(async (req: Request) => {
             stripe_subscription_id: subscription.id,
             period_end: periodEnd,
             pending_plan: null, // Clear any pending plan
+            payment_status: "ok", // Clear any payment failure state
           })
           .eq("organization_id", organizationId)
           .select();
@@ -273,17 +284,20 @@ Deno.serve(async (req: Request) => {
           break;
         }
 
-        // Check if this was a scheduled downgrade to free
+        // Check if this was a scheduled downgrade — validate against known plans
+        const validPlans = ["free", "pro", "team"];
         const downgradeTarget = subscription.metadata.downgrade_to;
+        const validatedPlan = (downgradeTarget && validPlans.includes(downgradeTarget)) ? downgradeTarget : "free";
 
         // Reset to free plan when subscription is cancelled
         const { error } = await supabase
           .from("organization_billing")
           .update({
-            plan: downgradeTarget || "free",
+            plan: validatedPlan,
             stripe_subscription_id: null,
             period_end: null,
             pending_plan: null,
+            payment_status: "ok",
           })
           .eq("organization_id", organizationId);
 
@@ -296,9 +310,7 @@ Deno.serve(async (req: Request) => {
         }
 
         console.log(
-          `Cancelled subscription for organization ${organizationId}, reset to ${
-            downgradeTarget || "free"
-          } plan`
+          `Cancelled subscription for organization ${organizationId}, reset to ${validatedPlan} plan`
         );
         break;
       }
@@ -476,10 +488,39 @@ Deno.serve(async (req: Request) => {
           `Updating period_end for organization ${organizationId} to: ${periodEnd}`
         );
 
+        // Determine invoice type so we only clear the matching payment_status.
+        // Plan invoice success clears 'past_due'; metered invoice success clears 'blocked'.
+        // This prevents a plan invoice success from incorrectly clearing a 'blocked' state
+        // caused by an unpaid metered invoice (and vice-versa).
+        const hasMeteredLineItems = invoice.lines?.data?.some(
+          (line) => {
+            const price = line.price as Stripe.Price;
+            return price?.type === "metered" || price?.recurring?.meter != null;
+          }
+        );
+        const isMeteredInvoice =
+          invoice.billing_reason === "subscription_threshold" || hasMeteredLineItems;
+
+        // Fetch current payment_status to decide whether to clear it
+        const { data: currentBilling } = await supabase
+          .from("organization_billing")
+          .select("payment_status")
+          .eq("organization_id", organizationId)
+          .maybeSingle();
+
+        const currentStatus = currentBilling?.payment_status || "ok";
+        let newPaymentStatus = currentStatus;
+        if (isMeteredInvoice && currentStatus === "blocked") {
+          newPaymentStatus = "ok";
+        } else if (!isMeteredInvoice && currentStatus === "past_due") {
+          newPaymentStatus = "ok";
+        }
+
         const { data, error } = await supabase
           .from("organization_billing")
           .update({
             period_end: periodEnd,
+            payment_status: newPaymentStatus,
           })
           .eq("organization_id", organizationId)
           .select();
@@ -493,7 +534,7 @@ Deno.serve(async (req: Request) => {
         }
 
         console.log(
-          `Payment succeeded for organization ${organizationId}, period_end updated to: ${periodEnd}`,
+          `Payment succeeded for organization ${organizationId}, period_end updated to: ${periodEnd}, payment_status: ${currentStatus} → ${newPaymentStatus} (${isMeteredInvoice ? "metered" : "plan"} invoice)`,
           data
         );
         break;
@@ -516,31 +557,75 @@ Deno.serve(async (req: Request) => {
           break;
         }
 
-        console.error(`Payment failed for organization ${organizationId}`);
+        console.error(`Payment failed for organization ${organizationId}, billing_reason: ${invoice.billing_reason}`);
 
-        // Check if this subscription was activated with a 100% off promo code
-        // and has no payment method (promo code expired scenario)
-        const wasActivatedWithPromo = subscription.metadata.activated_with_promo === "true";
-        const hasNoPaymentMethod = !subscription.default_payment_method;
+        // Determine if this is a metered usage invoice or a plan subscription invoice.
+        // Legacy metered prices have price.type === 'metered'.
+        // New Stripe Billing Meters have price.type === 'recurring' with price.recurring.meter set.
+        // billing_reason === 'subscription_threshold' also indicates metered usage.
+        const hasMeteredLineItems = invoice.lines?.data?.some(
+          (line) => {
+            const price = line.price as Stripe.Price;
+            return price?.type === "metered" || price?.recurring?.meter != null;
+          }
+        );
+        const isMeteredInvoice =
+          invoice.billing_reason === "subscription_threshold" || hasMeteredLineItems;
 
-        if (wasActivatedWithPromo && hasNoPaymentMethod) {
+        if (isMeteredInvoice) {
+          // Metered usage payment failed → hard block all writes
           console.log(
-            `Subscription ${subscriptionId} was activated with promo code and has no payment method. ` +
-            `Promo code likely expired. Scheduling downgrade to free plan.`
+            `Metered usage payment failed for organization ${organizationId}, setting payment_status to blocked`
           );
 
+          const { error } = await supabase
+            .from("organization_billing")
+            .update({
+              payment_status: "blocked",
+            })
+            .eq("organization_id", organizationId);
+
+          if (error) {
+            console.error(
+              `Error setting payment_status=blocked for organization ${organizationId}:`,
+              error
+            );
+          } else {
+            console.log(
+              `Set payment_status=blocked for organization ${organizationId} due to metered payment failure`
+            );
+          }
+        } else {
+          // Plan subscription payment failed → set past_due and schedule downgrade to free
+          console.log(
+            `Plan payment failed for organization ${organizationId}, setting payment_status to past_due and scheduling downgrade`
+          );
+
+          const { error: statusError } = await supabase
+            .from("organization_billing")
+            .update({
+              payment_status: "past_due",
+            })
+            .eq("organization_id", organizationId);
+
+          if (statusError) {
+            console.error(
+              `Error setting payment_status=past_due for organization ${organizationId}:`,
+              statusError
+            );
+          }
+
+          // Schedule downgrade to free plan
           try {
-            // Cancel subscription at period end (auto-downgrade to free)
             await stripe.subscriptions.update(subscriptionId, {
               cancel_at_period_end: true,
               metadata: {
                 ...subscription.metadata,
                 downgrade_to: "free",
-                downgrade_reason: "promo_expired_no_payment_method",
+                downgrade_reason: "plan_payment_failed",
               },
             });
 
-            // Update database to set pending_plan
             const { error } = await supabase
               .from("organization_billing")
               .update({
@@ -555,12 +640,12 @@ Deno.serve(async (req: Request) => {
               );
             } else {
               console.log(
-                `Set pending_plan to free for organization ${organizationId} due to promo expiration`
+                `Scheduled downgrade to free for organization ${organizationId} due to plan payment failure`
               );
             }
           } catch (err) {
             console.error(
-              `Error handling promo expiration for organization ${organizationId}:`,
+              `Error scheduling downgrade for organization ${organizationId}:`,
               err
             );
           }