fix(billing): fetch line items for checkout.session.completed webhooks

_extract_line_items() was returning [] for all checkout sessions, which
meant _handle_transaction_completed never processed credit packs, sticky
boosts, or business plan PDF purchases. Now fetches line items from the
Stripe API using the session ID, with a fallback to embedded line_items.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-03-05 10:49:41 +01:00
parent 6e3c5554aa
commit 66c2dfce66

View File

@@ -346,16 +346,30 @@ def _get_period_end(obj: dict) -> str | None:
def _extract_line_items(session_obj: dict) -> list[dict]:
"""Extract line items from a Checkout Session in Paddle-compatible format.
Stripe sessions don't embed line items directly — we'd need an extra API call.
For webhook handling, the key info (price_id) comes from subscription items.
Returns items in the format: [{"price": {"id": "price_xxx"}}]
Stripe doesn't embed line_items in checkout.session.completed webhooks,
so we fetch them via the API. Returns [{"price": {"id": "price_xxx"}}].
"""
# For checkout.session.completed, line_items aren't in the webhook payload.
# The webhook handler for subscription.activated fetches them separately.
# For one-time payments, we can reconstruct from the session's line_items
# via the Stripe API, but to keep webhook handling fast we skip this and
# handle it via the subscription events instead.
return []
session_id = session_obj.get("id", "")
if not session_id or not session_id.startswith("cs_"):
return []
try:
s = _stripe_client()
line_items = s.checkout.Session.list_line_items(session_id, limit=20)
return [
{"price": {"id": item["price"]["id"]}}
for item in line_items.get("data", [])
if item.get("price", {}).get("id")
]
except Exception:
logger.warning("Failed to fetch line_items for session %s", session_id)
# Fallback: check if line_items were embedded in the payload (e.g. tests)
embedded = session_obj.get("line_items", {}).get("data", [])
return [
{"price": {"id": item["price"]["id"]}}
for item in embedded
if item.get("price", {}).get("id")
]
def _extract_sub_items(sub_obj: dict) -> list[dict]: