Merge branch 'main' of https://github.com/LubomirGeorgiev/cloudflare-workers-nextjs-saas-template

Author: LubomirGeorgiev

src/actions/credits.action.ts

@@ -3,7 +3,7 @@
 import { requireVerifiedEmail } from "@/utils/auth";
 import {
   getCreditTransactions,
-  updateUserCredits,
+  addUserCredits,
   logTransaction,
   getCreditPackage,
 } from "@/utils/credits";
@@ -12,6 +12,7 @@ import { getStripe } from "@/lib/stripe";
 import { MAX_TRANSACTIONS_PER_PAGE, CREDITS_EXPIRATION_YEARS } from "@/constants";
 import ms from "ms";
 import { withRateLimit, RATE_LIMITS } from "@/utils/with-rate-limit";
+import { updateAllSessionsOfUser } from "@/utils/kv-session";
 
 // Action types
 type GetTransactionsInput = {
@@ -130,7 +131,7 @@ export async function confirmPayment({ packageId, paymentIntentId }: PurchaseCre
       }
 
       // Add credits and log transaction
-      await updateUserCredits(session.user.id, creditPackage.credits);
+      await addUserCredits(session.user.id, creditPackage.credits);
       await logTransaction({
         userId: session.user.id,
         amount: creditPackage.credits,
@@ -140,6 +141,9 @@ export async function confirmPayment({ packageId, paymentIntentId }: PurchaseCre
         paymentIntentId: paymentIntent?.id
       });
 
+      // Update all KV sessions to reflect the new credit balance
+      await updateAllSessionsOfUser(session.user.id);
+
       return { success: true };
     } catch (error) {
       console.error("Purchase error:", error);

src/utils/credits.ts

@@ -47,14 +47,25 @@ async function processExpiredCredits(userId: string, currentTime: Date) {
   // Process each expired transaction
   for (const transaction of expiredTransactions) {
     try {
-      // First, mark the transaction as processed to prevent double processing
-      await db
+      // Atomically mark the transaction as processed ONLY if it hasn't been processed yet
+      // This prevents race conditions where multiple requests try to process the same transaction
+      const updateResult = await db
         .update(creditTransactionTable)
         .set({
           expirationDateProcessedAt: currentTime,
           remainingAmount: 0, // All remaining credits are expired
         })
-        .where(eq(creditTransactionTable.id, transaction.id));
+        .where(and(
+          eq(creditTransactionTable.id, transaction.id),
+          isNull(creditTransactionTable.expirationDateProcessedAt),
+          eq(creditTransactionTable.remainingAmount, transaction.remainingAmount)
+        ))
+        .returning({ id: creditTransactionTable.id });
+
+      // If no rows were updated, another request already processed this transaction
+      if (!updateResult || updateResult.length === 0) {
+        continue;
+      }
 
       // Then deduct the expired credits from user's balance
       await db
@@ -70,27 +81,14 @@ async function processExpiredCredits(userId: string, currentTime: Date) {
   }
 }
 
-export async function updateUserCredits(userId: string, creditsToAdd: number) {
+export async function addUserCredits(userId: string, creditsToAdd: number) {
   const db = getDB();
   await db
     .update(userTable)
     .set({
       currentCredits: sql`${userTable.currentCredits} + ${creditsToAdd}`,
     })
     .where(eq(userTable.id, userId));
-
-  // Update all KV sessions to reflect the new credit balance
-  await updateAllSessionsOfUser(userId);
-}
-
-async function updateLastRefreshDate(userId: string, date: Date) {
-  const db = getDB();
-  await db
-    .update(userTable)
-    .set({
-      lastCreditRefreshAt: date,
-    })
-    .where(eq(userTable.id, userId));
 }
 
 export async function logTransaction({
@@ -140,14 +138,45 @@ export async function addFreeMonthlyCreditsIfNeeded(session: KVSession): Promise
       return user?.currentCredits ?? 0;
     }
 
+    // Calculate one month ago from current time (using calendar month logic)
+    const oneMonthAgo = new Date(currentTime);
+    oneMonthAgo.setMonth(oneMonthAgo.getMonth() - 1);
+
+    // Update last refresh date FIRST to act as a distributed lock
+    // This prevents race conditions where multiple requests try to add credits simultaneously
+    const updateResult = await db
+      .update(userTable)
+      .set({
+        lastCreditRefreshAt: currentTime,
+      })
+      .where(and(
+        eq(userTable.id, session.userId),
+        or(
+          isNull(userTable.lastCreditRefreshAt),
+          lt(userTable.lastCreditRefreshAt, oneMonthAgo) // More than 1 calendar month ago
+        )
+      ))
+      .returning({ lastCreditRefreshAt: userTable.lastCreditRefreshAt });
+
+    // If no rows were updated, another request already processed the refresh
+    if (!updateResult || updateResult.length === 0) {
+      const currentUser = await db.query.userTable.findFirst({
+        where: eq(userTable.id, session.userId),
+        columns: {
+          currentCredits: true,
+        },
+      });
+      return currentUser?.currentCredits ?? 0;
+    }
+
     // Process any expired credits first
     await processExpiredCredits(session.userId, currentTime);
 
     // Add free monthly credits with 1 month expiration
     const expirationDate = new Date(currentTime);
     expirationDate.setMonth(expirationDate.getMonth() + 1);
 
-    await updateUserCredits(session.userId, FREE_MONTHLY_CREDITS);
+    await addUserCredits(session.userId, FREE_MONTHLY_CREDITS);
     await logTransaction({
       userId: session.userId,
       amount: FREE_MONTHLY_CREDITS,
@@ -156,8 +185,8 @@ export async function addFreeMonthlyCreditsIfNeeded(session: KVSession): Promise
       expirationDate
     });
 
-    // Update last refresh date
-    await updateLastRefreshDate(session.userId, currentTime);
+    // Update all KV sessions to reflect the new credit balance and lastCreditRefreshAt
+    await updateAllSessionsOfUser(session.userId);
 
     // Get the updated credit balance from the database
     const updatedUser = await db.query.userTable.findFirst({
@@ -215,30 +244,57 @@ export async function consumeCredits({ userId, amount, description }: { userId:
   });
 
   let remainingToDeduct = amount;
+  let actuallyDeducted = 0;
 
   // Deduct from each transaction until we've deducted the full amount
   for (const transaction of activeTransactionsWithBalance) {
     if (remainingToDeduct <= 0) break;
 
     const deductFromThis = Math.min(transaction.remainingAmount, remainingToDeduct);
+    const newRemainingAmount = transaction.remainingAmount - deductFromThis;
 
-    await db
+    // Atomically update ONLY if the remainingAmount hasn't changed
+    // This prevents race conditions where multiple requests try to deduct from the same transaction
+    const updateResult = await db
       .update(creditTransactionTable)
       .set({
-        remainingAmount: transaction.remainingAmount - deductFromThis,
+        remainingAmount: newRemainingAmount,
       })
-      .where(eq(creditTransactionTable.id, transaction.id));
+      .where(and(
+        eq(creditTransactionTable.id, transaction.id),
+        eq(creditTransactionTable.remainingAmount, transaction.remainingAmount)
+      ))
+      .returning({ remainingAmount: creditTransactionTable.remainingAmount });
+
+    // If the update succeeded, count the deduction
+    if (updateResult && updateResult.length > 0) {
+      actuallyDeducted += deductFromThis;
+      remainingToDeduct -= deductFromThis;
+    }
+    // If update failed, another request modified this transaction, re-fetch and continue
+  }
 
-    remainingToDeduct -= deductFromThis;
+  // Verify we were able to deduct the full amount
+  if (actuallyDeducted < amount) {
+    throw new Error("Insufficient credits - concurrent modification detected");
   }
 
-  // Update total credits
-  await db
+  // Update total credits using SQL to ensure atomicity and prevent negative balance
+  const userUpdateResult = await db
     .update(userTable)
     .set({
       currentCredits: sql`${userTable.currentCredits} - ${amount}`,
     })
-    .where(eq(userTable.id, userId));
+    .where(and(
+      eq(userTable.id, userId),
+      sql`${userTable.currentCredits} >= ${amount}` // Ensure we don't go negative
+    ))
+    .returning({ currentCredits: userTable.currentCredits });
+
+  // If no rows were updated, we don't have enough credits (race condition)
+  if (!userUpdateResult || userUpdateResult.length === 0) {
+    throw new Error("Insufficient credits");
+  }
 
   // Log the usage transaction
   await db.insert(creditTransactionTable).values({
@@ -251,18 +307,10 @@ export async function consumeCredits({ userId, amount, description }: { userId:
     updatedAt: new Date(),
   });
 
-  // Get updated credit balance
-  const updatedUser = await db.query.userTable.findFirst({
-    where: eq(userTable.id, userId),
-    columns: {
-      currentCredits: true,
-    },
-  });
-
   // Update all KV sessions to reflect the new credit balance
   await updateAllSessionsOfUser(userId);
 
-  return updatedUser?.currentCredits ?? 0;
+  return userUpdateResult[0].currentCredits;
 }
 
 export async function getCreditTransactions({

src/utils/kv-session.ts

@@ -63,7 +63,7 @@ export interface KVSession {
  * IF YOU MAKE ANY CHANGES TO THE KVSESSION TYPE ABOVE, YOU NEED TO INCREMENT THIS VERSION.
  * THIS IS HOW WE TRACK WHEN WE NEED TO UPDATE THE SESSIONS IN THE KV STORE.
  */
-export const CURRENT_SESSION_VERSION = 2;
+export const CURRENT_SESSION_VERSION = 3;
 
 export async function getKV() {
   const { env } = getCloudflareContext();