Coverage for src/ai_lls_lib/payment/webhook_processor.py: 89%

1"""Stripe webhook event processing.""" 

2 

3import base64 

4import json 

5import logging 

6from datetime import UTC, datetime 

7from decimal import Decimal 

8from typing import Any, cast 

9 

10try: 

11 import stripe 

12 

13 HAS_STRIPE = True 

14except ImportError: 

15 stripe = None # type: ignore[assignment] 

16 HAS_STRIPE = False 

17 

18try: 

19 from botocore.exceptions import ClientError 

20 

21 HAS_BOTOCORE = True 

22except ImportError: 

23 HAS_BOTOCORE = False 

24 

25from .credit_manager import CreditManager 

26 

27logger = logging.getLogger(__name__) 

28 

29# Subscriptions grant no credits. An active subscription_status makes check_and_deduct 

30# return unlimited=True and skip deduction entirely, so any balance granted here could 

31# never be spent while the subscription is active (issue #134). 

32 

33 

34class WebhookProcessor: 

35 """Process Stripe webhook events.""" 

36 

37 def __init__(self, webhook_secret: str, credit_manager: CreditManager): 

38 """Initialize with webhook secret and credit manager.""" 

39 self.webhook_secret = webhook_secret 

40 self.credit_manager = credit_manager 

41 

42 def verify_and_parse(self, payload: str, signature: str) -> dict[str, Any]: 

43 """Verify webhook signature and parse event.""" 

44 if not HAS_STRIPE or not stripe: 

45 raise ImportError("stripe package not installed") 

46 

47 try: 

48 event = stripe.Webhook.construct_event(payload, signature, self.webhook_secret) 

49 # stripe 15 objects are not mappings: dict(event) raises TypeError. 

50 return cast(dict[str, Any], event.to_dict()) 

51 except ValueError as e: 

52 logger.error(f"Invalid webhook payload: {e}") 

53 raise 

54 except stripe.error.SignatureVerificationError as e: 

55 logger.error(f"Invalid webhook signature: {e}") 

56 raise 

57 

58 def process_event(self, event: dict[str, Any]) -> dict[str, Any]: 

59 """ 

60 Process a verified webhook event. 

61 Returns response data. 

62 """ 

63 event_type = event.get("type") 

64 event_data = event.get("data", {}).get("object", {}) 

65 

66 logger.info(f"Processing webhook event: {event_type}") 

67 

68 if event_type == "payment_intent.succeeded": 

69 return self._handle_payment_intent_succeeded(event_data) 

70 

71 elif event_type == "checkout.session.completed": 

72 return self._handle_checkout_completed(event_data) 

73 

74 elif event_type == "customer.subscription.created": 

75 return self._handle_subscription_created(event_data) 

76 

77 elif event_type == "customer.subscription.updated": 

78 return self._handle_subscription_updated(event_data) 

79 

80 elif event_type == "customer.subscription.deleted": 

81 return self._handle_subscription_deleted(event_data) 

82 

83 elif event_type == "invoice.payment_succeeded": 

84 return self._handle_invoice_paid(event_data) 

85 

86 elif event_type == "invoice.payment_failed": 

87 return self._handle_invoice_failed(event_data) 

88 

89 elif event_type == "charge.dispute.created": 

90 return self._handle_dispute_created(event_data) 

91 

92 else: 

93 logger.info(f"Unhandled event type: {event_type}") 

94 return {"message": f"Event {event_type} received but not processed"} 

95 

96 def process_eventbridge_event(self, event: dict[str, Any]) -> dict[str, Any]: 

97 """Process a Stripe event delivered via EventBridge. 

98 

99 EventBridge events are pre-verified by AWS, so no signature 

100 verification is needed. The Stripe event is in event["detail"]. 

101 Falls back to parsing event["body"] for direct webhook calls. 

102 """ 

103 detail = event.get("detail", {}) 

104 

105 if detail and "type" in detail: 

106 return self.process_event(detail) 

107 

108 # Fallback: direct webhook call with body 

109 body = event.get("body", "") 

110 if event.get("isBase64Encoded") and body: 

111 body = base64.b64decode(body).decode("utf-8") 

112 

113 if body: 

114 stripe_event = json.loads(body) 

115 return self.process_event(stripe_event) 

116 

117 logger.warning("EventBridge event has no detail or body") 

118 return {"message": "No event data found"} 

119 

120 def _handle_checkout_completed(self, session: dict[str, Any]) -> dict[str, Any]: 

121 """Handle successful checkout session for credit purchase.""" 

122 metadata = session.get("metadata", {}) 

123 user_id = metadata.get("user_id") 

124 

125 if not user_id: 

126 logger.error("No user_id in checkout session metadata") 

127 return {"error": "Missing user_id"} 

128 

129 if session.get("mode") == "payment": 

130 credits = int(metadata.get("credits", 0)) 

131 # Stripe delivers events at least once. Key the grant on the payment intent so a 

132 # redelivered session, or the matching payment_intent.succeeded event, cannot grant 

133 # the same purchase twice. Fall back to the session id if no intent is attached. 

134 idempotency_key = session.get("payment_intent") or session.get("id", "") 

135 

136 if credits > 0 and idempotency_key: 

137 granted = self.credit_manager.idempotent_credit_grant( 

138 user_id, credits, idempotency_key 

139 ) 

140 if not granted: 

141 logger.info(f"Checkout {idempotency_key} already processed for {user_id}") 

142 return {"message": "Checkout already processed, credits not added"} 

143 new_balance = self.credit_manager.get_balance(user_id) 

144 logger.info( 

145 f"Added {credits} credits to user {user_id}, new balance: {new_balance}" 

146 ) 

147 return {"credits_added": credits, "new_balance": new_balance} 

148 

149 return {"message": "Checkout processed"} 

150 

151 def _handle_subscription_created(self, subscription: dict[str, Any]) -> dict[str, Any]: 

152 """Handle new subscription creation: record state and the first billing period.""" 

153 metadata = subscription.get("metadata", {}) 

154 user_id = metadata.get("user_id") 

155 customer_id = subscription.get("customer") 

156 subscription_id = subscription.get("id") 

157 status = str(subscription.get("status", "unknown")) 

158 

159 if not user_id: 

160 logger.warning("No user_id in subscription metadata") 

161 return {"subscription_id": subscription_id, "status": status} 

162 

163 self.credit_manager.set_subscription_state( 

164 user_id=user_id, 

165 status=status, 

166 stripe_customer_id=customer_id, 

167 stripe_subscription_id=subscription_id, 

168 ) 

169 

170 logger.info(f"Created subscription {subscription_id} for user {user_id} ({status})") 

171 

172 # Record the initial billing period so renewals can be detected 

173 current_period_start = subscription.get("current_period_start") 

174 if current_period_start and self.credit_manager.table: 

175 try: 

176 self.credit_manager.table.update_item( 

177 Key={"user_id": user_id}, 

178 UpdateExpression="SET last_credited_period = :period", 

179 ExpressionAttributeValues={ 

180 ":period": Decimal(str(current_period_start)), 

181 }, 

182 ) 

183 except Exception as e: 

184 logger.error(f"Error setting last_credited_period for {user_id}: {e}") 

185 

186 return { 

187 "subscription_id": subscription_id, 

188 "status": status, 

189 } 

190 

191 def _handle_subscription_updated(self, subscription: dict[str, Any]) -> dict[str, Any]: 

192 """Handle subscription updates and record billing-period renewals. 

193 

194 When a billing period changes (detected via current_period_start) the new 

195 period is recorded with a conditional update so duplicate deliveries are inert. 

196 """ 

197 metadata = subscription.get("metadata", {}) 

198 user_id = metadata.get("user_id") 

199 subscription_id = subscription.get("id") 

200 status = str(subscription.get("status", "unknown")) 

201 

202 if not user_id: 

203 return {"subscription_id": subscription_id, "status": status} 

204 

205 # Always update subscription status 

206 self.credit_manager.set_subscription_state( 

207 user_id=user_id, status=status, stripe_subscription_id=subscription_id 

208 ) 

209 

210 renewed = False 

211 current_period_start = subscription.get("current_period_start") 

212 

213 if status == "active" and current_period_start and self.credit_manager.table: 

214 renewed = self._record_billing_period(user_id, current_period_start) 

215 

216 logger.info( 

217 f"Updated subscription {subscription_id} status to {status}" 

218 + (f", new billing period {current_period_start}" if renewed else "") 

219 ) 

220 

221 return { 

222 "subscription_id": subscription_id, 

223 "status": status, 

224 } 

225 

226 def _record_billing_period(self, user_id: str, current_period_start: int) -> bool: 

227 """Record a new billing period on the credits item. 

228 

229 Uses a conditional update on last_credited_period so duplicate webhook 

230 deliveries and concurrent handlers cannot advance it twice. 

231 

232 Returns True if a new period was recorded, False if already recorded. 

233 """ 

234 if not self.credit_manager.table: 

235 return False 

236 

237 try: 

238 # Read current last_credited_period 

239 response = self.credit_manager.table.get_item(Key={"user_id": user_id}) 

240 item = response.get("Item", {}) 

241 last_period = item.get("last_credited_period") 

242 

243 period_decimal = Decimal(str(current_period_start)) 

244 

245 # Check if this is a new period 

246 if last_period is not None and period_decimal <= Decimal(str(last_period)): 

247 logger.info(f"Period {current_period_start} already recorded for {user_id}") 

248 return False 

249 

250 # Conditional update to prevent race conditions 

251 if last_period is None: 

252 condition = "attribute_not_exists(last_credited_period)" 

253 expr_values = {":new_period": period_decimal} 

254 else: 

255 condition = "last_credited_period = :prev_period" 

256 expr_values = { 

257 ":new_period": period_decimal, 

258 ":prev_period": Decimal(str(last_period)), 

259 } 

260 

261 self.credit_manager.table.update_item( 

262 Key={"user_id": user_id}, 

263 UpdateExpression="SET last_credited_period = :new_period", 

264 ConditionExpression=condition, 

265 ExpressionAttributeValues=expr_values, 

266 ) 

267 

268 logger.info(f"Recorded billing period {current_period_start} for {user_id}") 

269 return True 

270 

271 except ClientError as e: 

272 if e.response["Error"]["Code"] == "ConditionalCheckFailedException": 

273 logger.info( 

274 f"Billing period {current_period_start} already recorded for {user_id} " 

275 "(concurrent delivery)" 

276 ) 

277 return False 

278 logger.error(f"Error recording billing period for {user_id}: {e}") 

279 raise 

280 except Exception as e: 

281 logger.error(f"Error recording billing period for {user_id}: {e}") 

282 return False 

283 

284 def _handle_subscription_deleted(self, subscription: dict[str, Any]) -> dict[str, Any]: 

285 """Handle subscription cancellation with timestamp.""" 

286 metadata = subscription.get("metadata", {}) 

287 user_id = metadata.get("user_id") 

288 subscription_id = subscription.get("id") 

289 

290 if user_id: 

291 self.credit_manager.set_subscription_state( 

292 user_id=user_id, status="cancelled", stripe_subscription_id=subscription_id 

293 ) 

294 

295 # Set cancellation timestamp 

296 if self.credit_manager.table: 

297 try: 

298 self.credit_manager.table.update_item( 

299 Key={"user_id": user_id}, 

300 UpdateExpression="SET subscription_cancelled_at = :now", 

301 ExpressionAttributeValues={ 

302 ":now": datetime.now(UTC).isoformat(), 

303 }, 

304 ) 

305 except Exception as e: 

306 logger.error(f"Error setting cancelled_at for {user_id}: {e}") 

307 

308 logger.info(f"Cancelled subscription {subscription_id} for user {user_id}") 

309 

310 return {"subscription_id": subscription_id, "status": "cancelled"} 

311 

312 def _handle_invoice_paid(self, invoice: dict[str, Any]) -> dict[str, Any]: 

313 """Handle successful subscription payment.""" 

314 customer_id = invoice.get("customer") 

315 amount = invoice.get("amount_paid", 0) / 100.0 

316 logger.info(f"Invoice paid: ${amount} from customer {customer_id}") 

317 return {"amount_paid": amount} 

318 

319 def _handle_invoice_failed(self, invoice: dict[str, Any]) -> dict[str, Any]: 

320 """Handle failed subscription payment.""" 

321 customer_id = invoice.get("customer") 

322 logger.warning(f"Invoice payment failed for customer {customer_id}") 

323 return {"status": "payment_failed"} 

324 

325 def _handle_payment_intent_succeeded(self, payment_intent: dict[str, Any]) -> dict[str, Any]: 

326 """Handle successful payment intent with idempotent credit grant.""" 

327 metadata = payment_intent.get("metadata", {}) 

328 user_id = metadata.get("user_id") 

329 

330 if not user_id: 

331 logger.error("No user_id in payment_intent metadata") 

332 return {"error": "Missing user_id"} 

333 

334 # Check if this is a verification charge ($1) 

335 if metadata.get("type") == "verification": 

336 logger.info(f"Verification charge completed for user {user_id}") 

337 return {"type": "verification", "status": "completed"} 

338 

339 # Get credits from metadata (set during payment creation) 

340 credits = int(metadata.get("credits", 0)) 

341 payment_intent_id = payment_intent.get("id", "") 

342 

343 if credits > 0 and payment_intent_id: 

344 granted = self.credit_manager.idempotent_credit_grant( 

345 user_id, credits, payment_intent_id 

346 ) 

347 if granted: 

348 new_balance = self.credit_manager.get_balance(user_id) 

349 logger.info( 

350 f"Added {credits} credits to user {user_id}, new balance: {new_balance}" 

351 ) 

352 return {"credits_added": credits, "new_balance": new_balance} 

353 else: 

354 logger.info(f"Payment {payment_intent_id} already processed for {user_id}") 

355 return {"message": "Payment already processed, credits not added"} 

356 

357 return {"message": "Payment processed"} 

358 

359 def _handle_dispute_created(self, dispute: dict[str, Any]) -> dict[str, Any]: 

360 """Handle charge dispute (mark account as disputed).""" 

361 charge_id = dispute.get("charge") 

362 

363 if not charge_id: 

364 logger.error("No charge_id in dispute") 

365 return {"error": "Missing charge_id"} 

366 

367 amount = dispute.get("amount", 0) / 100.0 

368 reason = dispute.get("reason", "unknown") 

369 

370 logger.warning(f"Dispute created for charge {charge_id}: ${amount}, reason: {reason}") 

371 

372 return {"dispute_id": dispute.get("id"), "status": "created", "amount": amount}