member-console/internal/integrations/stripe/workflows/webhook_invoice.go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo

package workflows

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"math"
	"time"

	"github.com/google/uuid"
	"go.temporal.io/sdk/temporal"

	"git.coopcloud.tech/wiki-cafe/member-console/internal/billing"
	internalstripe "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store"
)

// webhookInvoicePayload is the scrubbed payload stored for invoice.* events.
type webhookInvoicePayload struct {
	ID                   string `json:"id"`
	Customer             string `json:"customer"`
	Subscription         string `json:"subscription"`
	Status               string `json:"status"`
	AmountDue            int64  `json:"amount_due"`
	AmountPaid           int64  `json:"amount_paid"`
	Currency             string `json:"currency"`
	PeriodStart          int64  `json:"period_start"`
	PeriodEnd            int64  `json:"period_end"`
	DueDate              int64  `json:"due_date"`
	PaymentIntent        string `json:"payment_intent"`
	DefaultPaymentMethod string `json:"default_payment_method"`
	// Number is Stripe's own customer-facing invoice number (stripe-go's
	// Invoice.Number field), e.g. "A1B2C3D4-0001" -- an external reference
	// (invoice-numbers D5), recorded on the invoice mapping's
	// stripe_invoice_number, never the invoice's identity. Empty while the
	// invoice is a draft; Stripe assigns it once the invoice is finalized.
	// Not a PII key (internal/integrations/stripe/web/webhook.go), so it
	// survives ScrubPII unchanged.
	Number string              `json:"number"`
	Lines  invoicePayloadLines `json:"lines"`
	// Livemode is the invoice object's own flag, the environment the id
	// lives in (stripe-environment-stamp D1). It comes from the object
	// inside the stored payload, not from the event envelope, which this
	// activity never sees; every Stripe object carries it, so a parsed
	// payload always has a real answer here.
	Livemode bool `json:"livemode"`
}

type invoicePayloadLines struct {
	Data []invoiceLineItemPayload `json:"data"`
}

type invoiceLineItemPayload struct {
	ID               string                `json:"id"`
	SubscriptionItem string                `json:"subscription_item"`
	Price            invoicePriceRef       `json:"price"`
	Amount           int64                 `json:"amount"`
	Currency         string                `json:"currency"`
	Description      string                `json:"description"`
	Quantity         int64                 `json:"quantity"`
	Period           invoiceLineItemPeriod `json:"period"`
}

type invoicePriceRef struct {
	ID string `json:"id"`
}

type invoiceLineItemPeriod struct {
	Start int64 `json:"start"`
	End   int64 `json:"end"`
}

// handleInvoiceEvent dispatches invoice.* webhook events to sub-handlers.
func (a *WebhookActivities) handleInvoiceEvent(ctx context.Context, evt WebhookEvent) (string, error) {
	payload, err := a.readPayload(ctx, evt.ID)
	if err != nil {
		return "", err
	}

	var inv webhookInvoicePayload
	if err := json.Unmarshal(payload, &inv); err != nil {
		return "", fmt.Errorf("parse invoice payload: %w", err)
	}

	if inv.ID == "" {
		return "", fmt.Errorf("invoice payload missing id")
	}

	stripeQ := internalstripe.New(a.DB)

	switch evt.EventType {
	case "invoice.finalized":
		return a.handleInvoiceFinalized(ctx, evt, inv, stripeQ)
	case "invoice.paid":
		return a.handleInvoicePaid(ctx, evt, inv, stripeQ)
	case "invoice.payment_failed":
		return a.handleInvoicePaymentFailed(ctx, evt, inv, stripeQ)
	case "invoice.voided":
		return a.handleInvoiceVoided(ctx, evt, inv, stripeQ)
	default:
		a.Logger.Info("invoice sub-event skipped", slog.String("event_type", evt.EventType))
		return "skipped", nil
	}
}

func (a *WebhookActivities) handleInvoiceFinalized(ctx context.Context, evt WebhookEvent, inv webhookInvoicePayload, stripeQ *internalstripe.Queries) (string, error) {
	// Idempotency: skip if mapping already exists.
	_, err := stripeQ.GetInvoiceMappingByStripeID(ctx, sql.NullString{String: inv.ID, Valid: true})
	if err == nil {
		a.Logger.Info("invoice.finalized: mapping exists, skipping",
			slog.String("stripe_invoice_id", inv.ID))
		return "completed", nil
	}
	if err != sql.ErrNoRows {
		return "", fmt.Errorf("check existing invoice mapping: %w", err)
	}

	// Resolve billing_account_id from Stripe customer.
	custMapping, err := stripeQ.GetCustomerMappingByStripeCustomerID(ctx, sql.NullString{String: inv.Customer, Valid: true})
	if err != nil {
		return "", fmt.Errorf("resolve billing account from customer %s: %w", inv.Customer, err)
	}
	billingAccountID := custMapping.BillingAccountID

	// Resolve optional subscription_id.
	var subscriptionID uuid.NullUUID
	if inv.Subscription != "" {
		subMapping, err := stripeQ.GetSubscriptionMappingByStripeID(ctx, sql.NullString{String: inv.Subscription, Valid: true})
		if err != nil && err != sql.ErrNoRows {
			return "", fmt.Errorf("resolve subscription %s: %w", inv.Subscription, err)
		}
		if err == nil {
			subUUID, err := uuid.Parse(subMapping.SubscriptionID)
			if err != nil {
				return "", fmt.Errorf("parse subscription uuid: %w", err)
			}
			subscriptionID = uuid.NullUUID{UUID: subUUID, Valid: true}
		}
	}

	// Check the invoice-level amount before opening a transaction: nothing
	// has been written yet, so an out-of-range value can fail loudly with no
	// cleanup required (schema-hardening D6).
	amountDue, err := checkedInt32(evt.ProviderEventID, "amount_due", inv.AmountDue)
	if err != nil {
		return "", err
	}

	tx, err := a.DB.BeginTx(ctx, nil)
	if err != nil {
		return "", fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	billingQ := billing.New(tx)
	txStripeQ := internalstripe.New(tx)

	// Assign the platform's own reference number from the billing account's
	// transactional counter (invoice-numbers D1/D2): issuance for a
	// Stripe-originated invoice is this projection, so assignment happens
	// here, inside this transaction, never in the payload. A failure later
	// in this function rolls the whole transaction back (deferred
	// tx.Rollback above), which rewinds this UPDATE too -- the counter is
	// never burned on a failed projection.
	assigned, err := billingQ.AssignNextInvoiceNumber(ctx, billingAccountID)
	if err != nil {
		return "", fmt.Errorf("assign invoice number: %w", err)
	}
	invoiceNumber := fmt.Sprintf("%04d", assigned)

	// Create core invoice record with the assigned platform number.
	invoice, err := billingQ.CreateInvoice(ctx, billing.CreateInvoiceParams{
		BillingAccountID: billingAccountID,
		SubscriptionID:   subscriptionID,
		Status:           "open",
		AmountDue:        amountDue,
		AmountPaid:       0,
		Currency:         inv.Currency,
		PeriodStart:      nullTimeFromUnix(inv.PeriodStart),
		PeriodEnd:        nullTimeFromUnix(inv.PeriodEnd),
		DueDate:          nullTimeFromUnix(inv.DueDate),
		InvoiceNumber:    sql.NullString{String: invoiceNumber, Valid: true},
	})
	if err != nil {
		return "", fmt.Errorf("create invoice: %w", err)
	}

	// Create line items. A checked-conversion failure on any line item
	// returns before tx.Commit below, so the deferred Rollback discards the
	// invoice row above and any line items already inserted earlier in this
	// same uncommitted transaction — no partial row survives (relies on the
	// existing single-commit transaction rather than a separate up-front
	// validation pass; see schema-hardening tasks.md 4.1).
	for _, li := range inv.Lines.Data {
		// Resolve price_id from stripe price ID.
		priceMapping, err := txStripeQ.GetPriceMappingByStripeID(ctx, sql.NullString{String: li.Price.ID, Valid: true})
		if err != nil {
			a.Logger.Warn("invoice line item: price mapping not found, skipping line",
				slog.String("stripe_price_id", li.Price.ID),
				slog.String("stripe_invoice_id", inv.ID))
			continue
		}

		// Resolve optional subscription_item_id.
		var siID uuid.NullUUID
		if li.SubscriptionItem != "" {
			siMapping, err := txStripeQ.GetSubscriptionItemMappingByStripeID(ctx, sql.NullString{String: li.SubscriptionItem, Valid: true})
			if err == nil {
				siUUID, err := uuid.Parse(siMapping.SubscriptionItemID)
				if err == nil {
					siID = uuid.NullUUID{UUID: siUUID, Valid: true}
				}
			}
		}

		amount, err := checkedInt32(evt.ProviderEventID, fmt.Sprintf("lines[%s].amount", li.ID), li.Amount)
		if err != nil {
			return "", err
		}
		quantity, err := checkedInt32(evt.ProviderEventID, fmt.Sprintf("lines[%s].quantity", li.ID), li.Quantity)
		if err != nil {
			return "", err
		}

		_, err = billingQ.CreateInvoiceLineItem(ctx, billing.CreateInvoiceLineItemParams{
			InvoiceID:          invoice.InvoiceID,
			SubscriptionItemID: siID,
			PriceID:            priceMapping.PriceID,
			Amount:             amount,
			Currency:           li.Currency,
			Description:        sql.NullString{String: li.Description, Valid: li.Description != ""},
			Quantity:           quantity,
			PeriodStart:        nullTimeFromUnix(li.Period.Start),
			PeriodEnd:          nullTimeFromUnix(li.Period.End),
		})
		if err != nil {
			return "", fmt.Errorf("create invoice line item: %w", err)
		}
	}

	// Insert stripe mapping. stripe_invoice_number carries Stripe's own
	// customer-facing number (invoice-numbers D5) as received -- an
	// external reference, valid only when Stripe sent a non-empty one.
	_, err = txStripeQ.InsertInvoiceMapping(ctx, internalstripe.InsertInvoiceMappingParams{
		InvoiceID:           invoice.InvoiceID,
		StripeInvoiceID:     sql.NullString{String: inv.ID, Valid: true},
		StripeInvoiceNumber: sql.NullString{String: inv.Number, Valid: inv.Number != ""},
		SyncStatus:          "synced",
		Livemode:            sql.NullBool{Bool: inv.Livemode, Valid: true},
	})
	if err != nil {
		return "", fmt.Errorf("insert invoice mapping: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return "", fmt.Errorf("commit invoice.finalized: %w", err)
	}

	a.Logger.Info("invoice.finalized processed",
		slog.String("stripe_invoice_id", inv.ID),
		slog.String("invoice_id", invoice.InvoiceID))
	return "completed", nil
}

func (a *WebhookActivities) handleInvoicePaid(ctx context.Context, evt WebhookEvent, inv webhookInvoicePayload, stripeQ *internalstripe.Queries) (string, error) {
	// Resolve core invoice_id from stripe invoice mapping.
	invoiceMapping, err := stripeQ.GetInvoiceMappingByStripeID(ctx, sql.NullString{String: inv.ID, Valid: true})
	if err != nil {
		return "", fmt.Errorf("resolve invoice mapping for %s: %w", inv.ID, err)
	}

	// Idempotency. With a payment intent, the payment mapping is the record
	// of having handled it. Without one (a zero-amount invoice, one paid out
	// of band) the mapping is never written, so the guard is the invoice's
	// own succeeded payment; before it, a second invoice.paid for such an
	// invoice created a second payment row (2026-09 audit candidate).
	if inv.PaymentIntent != "" {
		_, err := stripeQ.GetPaymentMappingByStripePaymentIntentID(ctx, sql.NullString{String: inv.PaymentIntent, Valid: true})
		if err == nil {
			a.Logger.Info("invoice.paid: payment mapping exists, skipping",
				slog.String("stripe_payment_intent_id", inv.PaymentIntent))
			return "completed", nil
		}
		if err != sql.ErrNoRows {
			return "", fmt.Errorf("check existing payment mapping: %w", err)
		}
	} else {
		payments, err := billing.New(a.DB).GetPaymentsByInvoiceID(ctx, invoiceMapping.InvoiceID)
		if err != nil {
			return "", fmt.Errorf("check existing payments for invoice %s: %w", invoiceMapping.InvoiceID, err)
		}
		for _, existing := range payments {
			if existing.Status == "succeeded" {
				a.Logger.Info("invoice.paid: a succeeded payment already exists for the invoice, skipping",
					slog.String("invoice_id", invoiceMapping.InvoiceID), slog.String("payment_id", existing.PaymentID))
				return "completed", nil
			}
		}
	}

	// Resolve billing_account_id from customer mapping.
	custMapping, err := stripeQ.GetCustomerMappingByStripeCustomerID(ctx, sql.NullString{String: inv.Customer, Valid: true})
	if err != nil {
		return "", fmt.Errorf("resolve billing account from customer %s: %w", inv.Customer, err)
	}

	// Checked once, up front, before opening the transaction: amount_paid is
	// written to two rows below (the invoice and the payment) and must not
	// diverge, and checking before any write means an out-of-range value
	// fails with nothing to roll back (schema-hardening D6).
	amountPaid, err := checkedInt32(evt.ProviderEventID, "amount_paid", inv.AmountPaid)
	if err != nil {
		return "", err
	}

	tx, err := a.DB.BeginTx(ctx, nil)
	if err != nil {
		return "", fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	billingQ := billing.New(tx)
	txStripeQ := internalstripe.New(tx)

	// Mark invoice paid.
	now := time.Now()
	_, err = billingQ.UpdateInvoicePaid(ctx, billing.UpdateInvoicePaidParams{
		InvoiceID:  invoiceMapping.InvoiceID,
		AmountPaid: amountPaid,
		PaidAt:     sql.NullTime{Time: now, Valid: true},
	})
	if err != nil {
		return "", fmt.Errorf("update invoice paid: %w", err)
	}

	// invoice.paid never assigns or touches invoice_number (invoice-numbers
	// D2): assignment happens once, at issuance, in handleInvoiceFinalized.

	// Resolve optional payment method (already captured via payment_method.attached).
	var pmID uuid.NullUUID
	if inv.DefaultPaymentMethod != "" {
		pmMapping, err := txStripeQ.GetPaymentMethodMappingByStripeID(ctx, sql.NullString{String: inv.DefaultPaymentMethod, Valid: true})
		if err == nil {
			pmUUID, err := uuid.Parse(pmMapping.PaymentMethodID)
			if err == nil {
				pmID = uuid.NullUUID{UUID: pmUUID, Valid: true}
			}
		}
	}

	// Create payment record.
	payment, err := billingQ.CreatePayment(ctx, billing.CreatePaymentParams{
		InvoiceID:        invoiceMapping.InvoiceID,
		BillingAccountID: custMapping.BillingAccountID,
		PaymentMethodID:  pmID,
		Amount:           amountPaid,
		Currency:         inv.Currency,
		Status:           "succeeded",
		FailedAt:         sql.NullTime{},
	})
	if err != nil {
		return "", fmt.Errorf("create payment: %w", err)
	}

	// Insert payment mapping.
	if inv.PaymentIntent != "" {
		// The payment intent belongs to the invoice, so it lives in the
		// invoice's environment (stripe-environment-stamp D1).
		_, err = txStripeQ.InsertPaymentMapping(ctx, internalstripe.InsertPaymentMappingParams{
			PaymentID:             payment.PaymentID,
			StripePaymentIntentID: sql.NullString{String: inv.PaymentIntent, Valid: true},
			SyncStatus:            "synced",
			Livemode:              sql.NullBool{Bool: inv.Livemode, Valid: true},
		})
		if err != nil {
			return "", fmt.Errorf("insert payment mapping: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return "", fmt.Errorf("commit invoice.paid: %w", err)
	}

	a.Logger.Info("invoice.paid processed",
		slog.String("stripe_invoice_id", inv.ID),
		slog.String("payment_id", payment.PaymentID))
	return "completed", nil
}

func (a *WebhookActivities) handleInvoicePaymentFailed(ctx context.Context, evt WebhookEvent, inv webhookInvoicePayload, stripeQ *internalstripe.Queries) (string, error) {
	// Idempotency: skip if payment mapping already exists.
	if inv.PaymentIntent != "" {
		_, err := stripeQ.GetPaymentMappingByStripePaymentIntentID(ctx, sql.NullString{String: inv.PaymentIntent, Valid: true})
		if err == nil {
			a.Logger.Info("invoice.payment_failed: payment mapping exists, skipping",
				slog.String("stripe_payment_intent_id", inv.PaymentIntent))
			return "completed", nil
		}
		if err != sql.ErrNoRows {
			return "", fmt.Errorf("check existing payment mapping: %w", err)
		}
	}

	// Resolve core invoice_id.
	invoiceMapping, err := stripeQ.GetInvoiceMappingByStripeID(ctx, sql.NullString{String: inv.ID, Valid: true})
	if err != nil {
		return "", fmt.Errorf("resolve invoice mapping for %s: %w", inv.ID, err)
	}

	// Resolve billing_account_id.
	custMapping, err := stripeQ.GetCustomerMappingByStripeCustomerID(ctx, sql.NullString{String: inv.Customer, Valid: true})
	if err != nil {
		return "", fmt.Errorf("resolve billing account from customer %s: %w", inv.Customer, err)
	}

	// Checked up front, before opening the transaction, so an out-of-range
	// value fails with nothing to roll back (schema-hardening D6).
	amountDue, err := checkedInt32(evt.ProviderEventID, "amount_due", inv.AmountDue)
	if err != nil {
		return "", err
	}

	tx, err := a.DB.BeginTx(ctx, nil)
	if err != nil {
		return "", fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback() //nolint:errcheck

	billingQ := billing.New(tx)
	txStripeQ := internalstripe.New(tx)

	now := time.Now()
	payment, err := billingQ.CreatePayment(ctx, billing.CreatePaymentParams{
		InvoiceID:        invoiceMapping.InvoiceID,
		BillingAccountID: custMapping.BillingAccountID,
		Amount:           amountDue,
		Currency:         inv.Currency,
		Status:           "failed",
		FailedAt:         sql.NullTime{Time: now, Valid: true},
	})
	if err != nil {
		return "", fmt.Errorf("create failed payment: %w", err)
	}

	if inv.PaymentIntent != "" {
		// The payment intent belongs to the invoice, so it lives in the
		// invoice's environment (stripe-environment-stamp D1).
		_, err = txStripeQ.InsertPaymentMapping(ctx, internalstripe.InsertPaymentMappingParams{
			PaymentID:             payment.PaymentID,
			StripePaymentIntentID: sql.NullString{String: inv.PaymentIntent, Valid: true},
			SyncStatus:            "synced",
			Livemode:              sql.NullBool{Bool: inv.Livemode, Valid: true},
		})
		if err != nil {
			return "", fmt.Errorf("insert payment mapping: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return "", fmt.Errorf("commit invoice.payment_failed: %w", err)
	}

	a.Logger.Info("invoice.payment_failed processed",
		slog.String("stripe_invoice_id", inv.ID),
		slog.String("payment_id", payment.PaymentID))
	return "completed", nil
}

func (a *WebhookActivities) handleInvoiceVoided(ctx context.Context, evt WebhookEvent, inv webhookInvoicePayload, stripeQ *internalstripe.Queries) (string, error) {
	invoiceMapping, err := stripeQ.GetInvoiceMappingByStripeID(ctx, sql.NullString{String: inv.ID, Valid: true})
	if err == sql.ErrNoRows {
		a.Logger.Warn("invoice.voided: no mapping found, skipping",
			slog.String("stripe_invoice_id", inv.ID))
		return "completed", nil
	}
	if err != nil {
		return "", fmt.Errorf("resolve invoice mapping for %s: %w", inv.ID, err)
	}

	billingQ := billing.New(a.DB)
	now := time.Now()
	_, err = billingQ.UpdateInvoiceVoided(ctx, billing.UpdateInvoiceVoidedParams{
		InvoiceID: invoiceMapping.InvoiceID,
		VoidedAt:  sql.NullTime{Time: now, Valid: true},
	})
	if err != nil {
		return "", fmt.Errorf("update invoice voided: %w", err)
	}

	// invoice.voided never assigns or touches invoice_number (invoice-numbers
	// D2): a voided invoice keeps the number it was issued with.

	a.Logger.Info("invoice.voided processed", slog.String("stripe_invoice_id", inv.ID))
	return "completed", nil
}

// checkedInt32 converts a 64-bit Stripe amount or quantity into the 32-bit
// width the billing schema columns use. Stripe sends amounts in minor units
// (cents), and an ordinary invoice in a high-magnitude currency (e.g.
// Indonesian rupiah) can exceed math.MaxInt32; a bare int32(v) cast on such a
// value silently wraps to a negative number and gets stored behind a success
// log (schema-hardening D6). Out-of-range values fail loudly instead: the
// returned error is a non-retryable Temporal application error naming the
// webhook event, the field, and the offending value, so Temporal does not
// retry-spin on an event whose amount will never fit and no wrapped value is
// ever written. Never clamp.
func checkedInt32(providerEventID, field string, v int64) (int32, error) {
	if v < math.MinInt32 || v > math.MaxInt32 {
		return 0, temporal.NewNonRetryableApplicationError(
			fmt.Sprintf("stripe webhook event %s: field %s value %d is out of int32 range", providerEventID, field, v),
			"AmountOutOfRange",
			nil,
		)
	}
	return int32(v), nil
}

// nullTimeFromUnix converts a unix timestamp to sql.NullTime.
// A zero or negative value is treated as NULL.
func nullTimeFromUnix(ts int64) sql.NullTime {
	if ts <= 0 {
		return sql.NullTime{}
	}
	return sql.NullTime{Time: time.Unix(ts, 0), Valid: true}
}