Refactor Paddle customer handling to improve error extraction and logging for existing customers

Author: Bharani0012

src/backend/base/langflow/services/auth/clerk_utils.py

@@ -357,52 +357,51 @@ async def clerk_token_middleware(request: Request, call_next):
         return await call_next(request)
 
     ctx_token: Token | None = None
-    response = None
 
     try:
         # 1️⃣ Clerk JWT
         auth_header = request.headers.get("Authorization")
         if auth_header and auth_header.startswith("Bearer "):
             token = auth_header[len("Bearer ") :]
+
             try:
                 payload = await verify_clerk_token(token)
-                ctx_token = auth_header_ctx.set(payload)
-                response = await call_next(request)
             except Exception as exc:  # noqa: BLE001
-
                 logger.warning(f"[ClerkMiddleware] Failed to verify Clerk token: {exc}")
-                response = JSONResponse(
+                return JSONResponse(
                     status_code=HTTP_401_UNAUTHORIZED,
                     content={"detail": "Invalid Clerk token"},
                 )
 
-        # 2️⃣ API Key
-        elif (api_key_header := request.headers.get("x-api-key")):
+            ctx_token = auth_header_ctx.set(payload)
+            return await call_next(request)
 
+        # 2️⃣ API Key
+        api_key_header = request.headers.get("x-api-key")
+        if api_key_header:
             from langflow.services.auth.api_key_codec import decode_api_key
 
             decoded = decode_api_key(api_key_header)
             if not decoded.is_encoded or not decoded.organization_id:
-                logger.warning(f"[ClerkMiddleware] Invalid or unscoped API key: {api_key_header}")
-                response = JSONResponse(
+                logger.warning("[ClerkMiddleware] Invalid or unscoped API key")
+                return JSONResponse(
                     status_code=HTTP_401_UNAUTHORIZED,
                     content={"detail": "Invalid or unscoped API key"},
                 )
-            else:
-                context_payload = {
+
+            ctx_token = auth_header_ctx.set(
+                {
                     ORG_ID_KEY: decoded.organization_id,
                     CLERK_JWT_UUID_KEY: decoded.user_id,
                 }
-                ctx_token = auth_header_ctx.set(context_payload)
-                response = await call_next(request)
+            )
+            return await call_next(request)
 
-        # 3️⃣ No auth header → pass through
-        else:
-            response = await call_next(request)
+        # 3️⃣ No auth → pass through
+        return await call_next(request)
 
     finally:
         if ctx_token is not None:
             auth_header_ctx.reset(ctx_token)
         else:
             auth_header_ctx.set(None)
-    return response

src/backend/base/langflow/services/paddle/subscriptions.py

@@ -5,6 +5,7 @@
 import asyncio
 from dataclasses import dataclass
 from datetime import datetime
+import re
 from typing import Any, Literal
 
 from lfx.log.logger import logger
@@ -28,6 +29,8 @@
 )
 from langflow.services.paddle.client import get_paddle_client
 
+_PADDLE_CUSTOMER_ID_RE = re.compile(r"(ctm_[a-z0-9]+)", re.IGNORECASE)
+
 
 async def ensure_paddle_customer_for_user(
     *,
@@ -56,6 +59,11 @@ async def _get_paddle_customer_id() -> str | None:
     return await get_paddle_customer_id_from_clerk_payload()
 
 
+def _extract_customer_id_from_error(exc: Exception) -> str | None:
+    match = _PADDLE_CUSTOMER_ID_RE.search(str(exc))
+    return match.group(1) if match else None
+
+
 async def _create_paddle_customer_and_update_clerk_metadata(
     client: Client,
     clerk_user_id: str,
@@ -76,35 +84,55 @@ async def _create_paddle_customer_and_update_clerk_metadata(
                         PADDLE_CUSTOM_DATA_USER_ID_KEY: str(clerk_user_id),
                     }
                 ),
+            ),
         )
-     )
         logger.info(f"Paddle customer creation response: {customer}")
     except Exception as exc:  # noqa: BLE001
-        if not _is_customer_already_exists_error(exc):
-            raise
-        logger.info(
-            "Paddle customer already exists for email %s; resolving existing customer.",
-            email,
-        )
-        existing_customer = await _find_existing_paddle_customer(
-            client=client,
-            email=email,
-            clerk_user_id=clerk_user_id,
-        )
-        if existing_customer is None:
-            logger.exception("Unable to resolve existing Paddle customer for email %s.", email)
-            raise
+        customer_id = _extract_customer_id_from_error(exc)
+
+        if not customer_id:
+            if not _is_customer_already_exists_error(exc):
+                raise
+
+            logger.info(
+                "Paddle customer already exists for email %s; resolving existing customer.",
+                email,
+            )
+            customer = await _find_existing_paddle_customer(
+                client=client,
+                email=email,
+                clerk_user_id=clerk_user_id,
+            )
+            if customer is None:
+                logger.exception("Unable to resolve existing Paddle customer for email %s.", email)
+                raise
+        else:
+            logger.info(
+                "Paddle reports existing customer %s for email %s",
+                customer_id,
+                email,
+            )
+            try:
+                customer = await asyncio.to_thread(client.customers.get, customer_id)
+            except Exception as exc:  # noqa: BLE001
+                logger.exception(
+                    "Failed to retrieve existing Paddle customer with ID %s for clerk user %s.",
+                    customer_id,
+                    clerk_user_id,
+                )
+                raise
+
         await _sync_paddle_customer_metadata(
             client=client,
-            customer=existing_customer,
+            customer=customer,
             clerk_user_id=clerk_user_id,
         )
-        paddle_customer_id = existing_customer.id
+
         await update_clerk_public_metadata(
             clerk_user_id=clerk_user_id,
-            public_metadata={PADDLE_CUSTOMER_ID_KEY: paddle_customer_id},
+            public_metadata={PADDLE_CUSTOMER_ID_KEY: customer.id},
         )
-        return paddle_customer_id
+        return customer.id
 
     paddle_customer_id = customer.id
 
@@ -119,7 +147,12 @@ async def _create_paddle_customer_and_update_clerk_metadata(
 
 def _is_customer_already_exists_error(exc: Exception) -> bool:
     message = str(exc).lower()
-    return "customer already exists" in message or ("already exists" in message and "customer" in message)
+    return (
+        "customer already exists" in message
+        or ("already exists" in message and "customer" in message)
+        or ("customer email conflicts" in message)
+        or ("email conflicts" in message and "customer" in message)
+    )
 
 
 async def _find_existing_paddle_customer(
@@ -159,8 +192,16 @@ async def _sync_paddle_customer_metadata(
 
     updated_custom_data = dict(existing_custom_data)
     updated_custom_data[PADDLE_CUSTOM_DATA_USER_ID_KEY] = str(clerk_user_id)
-    await asyncio.to_thread(
-        client.customers.update,
-        customer.id,
-        UpdateCustomer(custom_data=CustomData(updated_custom_data)),
-    )
+    try:
+        await asyncio.to_thread(
+            client.customers.update,
+            customer.id,
+            UpdateCustomer(custom_data=CustomData(updated_custom_data)),
+        )
+    except Exception as exc:  # noqa: BLE001
+        logger.exception(
+            "Failed to sync Paddle customer metadata for customer %s and clerk user %s.",
+            customer.id,
+            clerk_user_id,
+        )
+        raise