package hyperswitch import ( "context" "errors" "fmt" "strings" "time" "go.uber.org/zap" "gorm.io/gorm" "veza-backend-api/internal/core/marketplace" "veza-backend-api/internal/core/subscription" ) // SubscriptionWebhookProcessor handles Hyperswitch webhooks that target // subscription invoices (v1.0.9 item G Phase 2 — closes the // pending_payment state machine opened in Phase 1). // // State transitions: // - pending_payment + status=succeeded → invoice paid, sub active // - pending_payment + status=failed → invoice failed, sub expired // - already terminal → idempotent no-op (Hyperswitch // re-emits webhooks until 200 OK; we must accept replays) // - payment_id not in subscription_invoices → marketplace.ErrNotASubscription // (caller falls through to order webhook flow) // // The processor only flips a subscription out of pending_payment. // Subscriptions that have already transitioned (concurrent flow, manual // admin action, plan upgrade) are left alone — the invoice still gets // the terminal status update so the audit trail is consistent. type SubscriptionWebhookProcessor struct { db *gorm.DB logger *zap.Logger } // NewSubscriptionWebhookProcessor constructs a processor bound to the // given DB. The logger is optional; nil is replaced by zap.NewNop so the // happy path doesn't crash on a forgotten DI wire-up. func NewSubscriptionWebhookProcessor(db *gorm.DB, logger *zap.Logger) *SubscriptionWebhookProcessor { if logger == nil { logger = zap.NewNop() } return &SubscriptionWebhookProcessor{db: db, logger: logger} } // IsSubscriptionEventType returns true if the event_type matches the // "subscription.*" prefix used by Item G. The dispatcher in // marketplace.ProcessPaymentWebhook does NOT rely on this — it always // tries the invoice lookup first because the PSP can re-emit subscription // payments with payment_intent.* event types after a recovery flow. The // helper is exposed for callers that want to short-circuit the DB hit on // clearly non-subscription events. func IsSubscriptionEventType(eventType string) bool { return strings.HasPrefix(strings.ToLower(eventType), "subscription.") } // ProcessSubscriptionPayment satisfies marketplace.SubscriptionWebhookHandler. // Returns marketplace.ErrNotASubscription when the payment_id is not // associated with any subscription invoice — the caller treats that as // "not a subscription event" and falls through to the order flow. func (p *SubscriptionWebhookProcessor) ProcessSubscriptionPayment(ctx context.Context, paymentID, status string) error { if paymentID == "" { return errors.New("empty payment_id") } var inv subscription.Invoice if err := p.db.WithContext(ctx). Where("hyperswitch_payment_id = ?", paymentID). First(&inv).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return marketplace.ErrNotASubscription } return fmt.Errorf("lookup subscription invoice: %w", err) } var sub subscription.UserSubscription if err := p.db.WithContext(ctx). Where("id = ?", inv.SubscriptionID). First(&sub).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { // Invoice exists but subscription is gone — data integrity // issue worth surfacing rather than swallowing. return fmt.Errorf("subscription %s not found for invoice %s", inv.SubscriptionID, inv.ID) } return fmt.Errorf("lookup subscription: %w", err) } switch strings.ToLower(status) { case "succeeded": if sub.Status == subscription.StatusActive && inv.Status == subscription.InvoicePaid { p.logger.Debug("Subscription webhook: already active+paid (idempotent replay)", zap.String("subscription_id", sub.ID.String()), zap.String("payment_id", paymentID)) return nil } return p.activate(ctx, &inv, &sub, paymentID) case "failed": if sub.Status == subscription.StatusExpired && inv.Status == subscription.InvoiceFailed { p.logger.Debug("Subscription webhook: already expired+failed (idempotent replay)", zap.String("subscription_id", sub.ID.String()), zap.String("payment_id", paymentID)) return nil } return p.expire(ctx, &inv, &sub, paymentID) default: // Intermediate / unknown statuses (processing, requires_*) — log // and noop. Hyperswitch retries until a terminal status is acked, // so the transient ones are safe to ignore. p.logger.Debug("Subscription webhook: ignoring non-terminal status", zap.String("status", status), zap.String("payment_id", paymentID), zap.String("subscription_id", sub.ID.String())) return nil } } func (p *SubscriptionWebhookProcessor) activate(ctx context.Context, inv *subscription.Invoice, sub *subscription.UserSubscription, paymentID string) error { return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { now := time.Now() if err := tx.Model(&subscription.Invoice{}). Where("id = ?", inv.ID). Updates(map[string]any{ "status": subscription.InvoicePaid, "paid_at": now, }).Error; err != nil { return fmt.Errorf("update invoice: %w", err) } // Only flip the subscription if it's currently pending_payment. // A row already in active/canceled/upgraded must not be stomped // by a delayed webhook arrival. result := tx.Model(&subscription.UserSubscription{}). Where("id = ? AND status = ?", sub.ID, subscription.StatusPendingPayment). Updates(map[string]any{"status": subscription.StatusActive}) if result.Error != nil { return fmt.Errorf("update subscription: %w", result.Error) } if result.RowsAffected == 0 { p.logger.Warn("Subscription webhook: subscription not in pending_payment, invoice still flipped to paid", zap.String("subscription_id", sub.ID.String()), zap.String("current_status", string(sub.Status)), zap.String("payment_id", paymentID)) return nil } p.logger.Info("Subscription activated via Hyperswitch webhook", zap.String("subscription_id", sub.ID.String()), zap.String("invoice_id", inv.ID.String()), zap.String("payment_id", paymentID)) return nil }) } func (p *SubscriptionWebhookProcessor) expire(ctx context.Context, inv *subscription.Invoice, sub *subscription.UserSubscription, paymentID string) error { return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { if err := tx.Model(&subscription.Invoice{}). Where("id = ?", inv.ID). Updates(map[string]any{"status": subscription.InvoiceFailed}).Error; err != nil { return fmt.Errorf("update invoice: %w", err) } result := tx.Model(&subscription.UserSubscription{}). Where("id = ? AND status = ?", sub.ID, subscription.StatusPendingPayment). Updates(map[string]any{"status": subscription.StatusExpired}) if result.Error != nil { return fmt.Errorf("update subscription: %w", result.Error) } if result.RowsAffected == 0 { p.logger.Warn("Subscription webhook: subscription not in pending_payment, invoice still flipped to failed", zap.String("subscription_id", sub.ID.String()), zap.String("current_status", string(sub.Status)), zap.String("payment_id", paymentID)) return nil } p.logger.Info("Subscription expired via Hyperswitch webhook", zap.String("subscription_id", sub.ID.String()), zap.String("invoice_id", inv.ID.String()), zap.String("payment_id", paymentID)) return nil }) }