package workers

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"runtime/debug"
	"sort"
	"strings"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/shopspring/decimal"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/forms"
	"github.com/pingcap/dbaas/billing2/handler/centralhandler"
	"github.com/pingcap/dbaas/billing2/handler/paymenthandler"
	"github.com/pingcap/dbaas/billing2/model/payment"
	"github.com/pingcap/dbaas/billing2/model/payment/paymentsub"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"go.uber.org/zap"
)

// PaidItem is one outcome per (tenant, billing_config).
type PaidItem struct {
	TenantID         uint            `json:"tenant_id"`
	BillingConfigID  uint            `json:"billing_config_id"`
	MergedInvoiceIDs []uint          `json:"merged_invoice_ids"`
	Amount           decimal.Decimal `json:"amount"`
	PushType         string          `json:"push_type"`
	ChargeMode       string          `json:"charge_mode,omitempty"`
	StripeInvoiceID  string          `json:"stripe_invoice_id,omitempty"`
}

// PayingResult is the JSON written to stripe_invoice_paying.result on completion.
// Schema matches the GET /paying/{id} response in the design doc.
type PayingResult struct {
	Confirmed        []PaidItem `json:"confirmed"`
	CarryoverUpdated []PaidItem `json:"carryover_updated"`
	Expired          []PaidItem `json:"expired"`
	Skipped          []PaidItem `json:"skipped"`
	// Suspended lists invoices the worker parked in SUSPENDED because their
	// active payment method is managed_by_sales and their type is not
	// COMMITMENT. Sales handles billing out-of-band for those customers, so
	// the auto-push pipeline must not touch them.
	Suspended []PaidItem `json:"suspended"`
}

func payingResultJSON(r PayingResult) (string, error) {
	b, err := json.Marshal(r)
	if err != nil {
		return "", err
	}
	return string(b), nil
}

// ErrPayingNotRunning is returned by runPaying when the heartbeat probe finds
// the paying row no longer in 'running' (i.e. the admin or janitor force-failed
// it mid-run). The entry point treats it as a benign terminal state instead of
// failing the paying again.
var ErrPayingNotRunning = errors.New("paying is no longer running")

// Worker metrics. Namespace: dbaas_billing_stripe_invoice_paying_*.
var (
	payingTotal = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "dbaas",
		Subsystem: "billing",
		Name:      "stripe_invoice_paying_total",
	}, []string{"result"})

	payingPendingAge = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: "dbaas",
		Subsystem: "billing",
		Name:      "stripe_invoice_paying_pending_age_seconds",
	})

	payingDuration = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "dbaas",
		Subsystem: "billing",
		Name:      "stripe_invoice_paying_duration_seconds",
	})
)

func payingMetricInc(result string) { payingTotal.WithLabelValues(result).Inc() }

// monthRange parses "yyyy-mm" into [start, end) UTC bounds.
func monthRange(month string) (time.Time, time.Time, error) {
	start, err := time.Parse("2006-01", month)
	if err != nil {
		return time.Time{}, time.Time{}, fmt.Errorf("parse month %q: %w", month, err)
	}
	end := start.AddDate(0, 1, 0)
	return start, end, nil
}

// sameMonth reports whether all invoices in the slice fall in the same calendar
// month (used for the single-month auto-charge rule).
func sameMonth(invoices []forms.Invoice) bool {
	if len(invoices) == 0 {
		return false
	}
	first := invoices[0].StartDate.Month()
	firstYear := invoices[0].StartDate.Year()
	for _, inv := range invoices[1:] {
		if inv.StartDate.Month() != first || inv.StartDate.Year() != firstYear {
			return false
		}
	}
	return true
}

// tenantKindIsInternal fetches the tenant and returns whether its kind is internal.
// Network errors default to non-internal so the rest of the scan can proceed.
func tenantKindIsInternal(ctx context.Context, tenantID uint) bool {
	_ = ctx
	t, err := centralhandler.GetCentralHandler().GetTenant(tenantID)
	if err != nil || t == nil {
		return false
	}
	return t.Kind == forms.TenantKindInternal
}

// tenantIsPermanentFree reports whether the tenant currently hits the
// billing_permenent_free_orgs feature flag. Defense-in-depth for the paying
// worker: upstream CreateInvoiceWorker already creates these tenants' invoices
// in SUSPENDED status (so the GENERATED filter in ListStripeEligibleInvoices
// would naturally exclude them), but a feature-flag misconfiguration or
// race during invoice generation could leave a GENERATED row behind. We fail
// open on flag-check errors so a transient central-service hiccup doesn't
// freeze the entire paying run — the status filter is the primary safety net.
func tenantIsPermanentFree(ctx context.Context, tenantID uint) bool {
	_ = ctx
	hit, err := centralhandler.IsHitFeatureFlag(uint64(tenantID), billingPermanentFreeOrgsFeatureFlag)
	if err != nil {
		log.Warn("permanent-free feature flag check failed; treating as non-permanent-free",
			zap.Uint("tenantID", tenantID),
			zap.Error(err))
		return false
	}
	return hit
}

// StripeInvoicePayingWorker is the gocron entry point. Picks the oldest pending
// paying and runs the scan-and-push cycle.
func StripeInvoicePayingWorker() {
	// Per-tick marker: a one-line log at INFO so dev can grep
	// "StripeInvoicePayingWorker tick" and compute the actual cadence
	// from the gap between timestamps. The wrapworker also logs
	// "Worker Task Finished" with the Go function name, but a
	// dedicated marker here is easier to filter on.
	log.Info("StripeInvoicePayingWorker tick")
	payingModel := &payment.PayingModel{}
	invoiceModel := &payment.InvoiceModel{}

	paying, err := payingModel.ClaimNextPendingPaying()
	if err != nil {
		log.Info("no pending paying to claim", zap.Error(err))
		return
	}
	if paying == nil {
		return
	}
	log.Info("claimed paying",
		zap.Uint("payingID", paying.ID),
		zap.String("month", paying.Month),
		zap.Time("pickedUpAt", derefTime(paying.PickedUpAt)))
	if paying.PickedUpAt != nil {
		payingPendingAge.Set(time.Since(*paying.PickedUpAt).Seconds())
	}
	start := time.Now()
	defer func() { payingDuration.Observe(time.Since(start).Seconds()) }()

	// Catch any panic downstream (scan, push, DB) so the paying row is
	// cleanly marked failed instead of leaking a running row that the
	// janitor has to reap. defer is LIFO so this runs *after* the duration
	// observer defer above.
	defer func() {
		if r := recover(); r != nil {
			stack := string(debug.Stack())
			log.Error("paying worker panicked", zap.Any("recover", r), zap.String("stack", stack))
			failPaying(payingModel, paying, fmt.Sprintf("panic: %v\n%s", r, stack))
			payingMetricInc("failed")
		}
	}()

	startTime, endTime, err := monthRange(paying.Month)
	if err != nil {
		failPaying(payingModel, paying, fmt.Sprintf("invalid month on paying %d: %v", paying.ID, err))
		payingMetricInc("failed")
		return
	}

	ctx := context.Background()
	result, runErr := runPaying(ctx, paying, startTime, endTime, invoiceModel, payingModel)
	if runErr != nil {
		// H1 heartbeat: when the paying was force-failed / terminalised while
		// the run was still in flight, leave the terminal state intact instead
		// of failing it a second time (mirrors the already_terminal handling in
		// the CompletePaying path below).
		if errors.Is(runErr, ErrPayingNotRunning) {
			log.Info("paying was terminalised before run finished; leaving intact",
				zap.Uint("payingID", paying.ID))
			payingMetricInc("already_terminal")
			return
		}
		failPaying(payingModel, paying, runErr.Error())
		payingMetricInc("failed")
		return
	}

	resultJSON, err := payingResultJSON(result)
	if err != nil {
		failPaying(payingModel, paying, fmt.Sprintf("marshal result: %v", err))
		payingMetricInc("failed")
		return
	}
	if err := payingModel.CompletePaying(paying, resultJSON); err != nil {
		// M1 guard: CompletePaying returns ObjectNotExistsError when the row is
		// no longer in 'running' — i.e. the janitor / admin already flipped the
		// paying to a terminal state. Don't try FailPaying here: it would also
		// fail with ObjectNotExistsError and emit misleading "failed to mark
		// paying failed" log lines. The paying is in its intended terminal
		// state; we just log and surface as a separate metric label.
		if errors.Is(err, billing_errors.ObjectNotExistsError) {
			log.Info("paying was force-failed before completion; leaving terminal state intact",
				zap.Uint("payingID", paying.ID),
				zap.Error(err))
			payingMetricInc("already_terminal")
			return
		}
		log.Error("failed to mark paying complete", zap.Uint("payingID", paying.ID), zap.Error(err))
		failPaying(payingModel, paying, fmt.Sprintf("complete paying: %v", err))
		payingMetricInc("failed")
		return
	}
	payingMetricInc("completed")
	log.Info("paying completed",
		zap.Uint("payingID", paying.ID),
		zap.String("month", paying.Month),
		zap.Int("confirmed", len(result.Confirmed)),
		zap.Int("carryoverUpdated", len(result.CarryoverUpdated)),
		zap.Int("expired", len(result.Expired)),
	)
}

func failPaying(payingModel *payment.PayingModel, paying *entity.StripeInvoicePaying, msg string) {
	if err := payingModel.FailPaying(paying, msg); err != nil {
		log.Error("failed to mark paying failed", zap.Uint("payingID", paying.ID), zap.Error(err))
	}
	log.Error("paying failed", zap.Uint("payingID", paying.ID), zap.String("error", msg))
}

// runPaying is the scan + per-group decision. Split out so tests can drive it
// with a pinned time range without claiming.
//
// Carryover tracking is fully status-driven (see ListCarryoverInvoices):
// billing2_invoices.status = 'below_threshold_carry' is the single source
// of truth. The stripe_invoice_push_carryover pool table is NOT read or
// written here — no GetOrCreate, no UpsertWithAmount, no ClearSourceInvoices.
func runPaying(
	ctx context.Context,
	paying *entity.StripeInvoicePaying,
	start, end time.Time,
	invoiceModel *payment.InvoiceModel,
	payingModel *payment.PayingModel,
) (PayingResult, error) {
	// payingModel is used inline below for per-invoice Heartbeat probes; it
	// stayed on the signature so the worker entry point can swap model
	// implementations for testing without touching call sites.
	minAmount, err := decimal.NewFromString(config.Get().GetString(config.StripePushMinAmountUSD))
	if err != nil || !minAmount.IsPositive() {
		minAmount = decimal.NewFromFloat(1.00)
	}
	maxCarryoverMonths := config.Get().GetInt(config.StripePushMaxCarryoverMonths)
	if maxCarryoverMonths <= 0 {
		maxCarryoverMonths = 12
	}
	expiryCutoff := start.AddDate(0, -maxCarryoverMonths, 0)

	result := PayingResult{}
	log.Info("runPaying start",
		zap.Uint("payingID", paying.ID),
		zap.String("month", paying.Month),
		zap.Time("windowStart", start),
		zap.Time("windowEnd", end),
		zap.String("minAmount", minAmount.String()),
		zap.Int("maxCarryoverMonths", maxCarryoverMonths))

	// Pull all Stripe generated invoices in the month window.
	invoices, err := invoiceModel.ListStripeEligibleInvoices(start, end)
	if err != nil {
		return result, fmt.Errorf("scan invoices: %w", err)
	}
	log.Info("scanned eligible invoices",
		zap.Uint("payingID", paying.ID),
		zap.Int("count", len(invoices)),
		zap.Time("windowStart", start),
		zap.Time("windowEnd", end))

	// Group by (tenant_id, billing_config_id).
	type groupKey struct{ TenantID, BillingConfigID uint }
	groups := map[groupKey][]forms.Invoice{}
	for _, inv := range invoices {
		bc := uint(0)
		if inv.BindedBillingConfigID != nil {
			bc = *inv.BindedBillingConfigID
		}
		if bc == 0 {
			continue
		}
		k := groupKey{inv.TenantID, bc}
		groups[k] = append(groups[k], inv)
	}

	for key, inMonthInvoices := range groups {
		// Defense-in-depth: skip internal tenants.
		if tenantKindIsInternal(ctx, key.TenantID) {
			continue
		}

		// Defense-in-depth: skip tenants flagged as permanent-free.
		// CreateInvoiceWorker parks their invoices in SUSPENDED status, but
		// we re-check here so a flag-config error or generation-time race
		// can't accidentally push a permanent-free tenant's invoice to
		// Stripe. Mirrors the internal-tenant filter above.
		if tenantIsPermanentFree(ctx, key.TenantID) {
			continue
		}

		// Phase 4 of the invoice/PM decoupling: pull the active PM up front
		// so we can route managed-by-sales non-commitment invoices to
		// SUSPENDED before any push/expire/carry arithmetic runs.
		//
		// `managed_by_sales` flags customers whose billing is handled by the
		// sales team out-of-band (typically enterprise contracts); for them
		// the auto-push pipeline should only run on COMMITMENT invoices
		// (the actual contractual commitment fee). All other invoice types
		// (monthly / daily / hourly / inmonth / cost_quota / recal /
		// subordinated) get parked in SUSPENDED so the dashboard surfaces
		// them and a human can take over.
		pmForGroup, pmErr := (&paymentsub.PaymentModel{}).GetActivePaymentMethod(key.TenantID, key.BillingConfigID)
		if pmErr != nil {
			return result, fmt.Errorf("resolve active payment method for managed-by-sales check: %w", pmErr)
		}

		// Carryover source of truth: billing2_invoices.status. Invoices in
		// below_threshold_carry from previous months are the carried batch
		// awaiting the next merged push. The stripe_invoice_push_carryover
		// pool table is NOT consulted here (and no longer written by this
		// worker).
		carryInvoices, err := invoiceModel.ListCarryoverInvoices(key.TenantID, key.BillingConfigID)
		if err != nil {
			return result, fmt.Errorf("list carryover invoices: %w", err)
		}

		// Expiry: carried invoices older than the cutoff expire and drop out
		// of the pipeline; the rest keep waiting.
		expiredInvoiceIDs := []uint{}
		keptCarry := make([]forms.Invoice, 0, len(carryInvoices))
		for _, inv := range carryInvoices {
			if inv.StartDate.Before(expiryCutoff) {
				expiredInvoiceIDs = append(expiredInvoiceIDs, uint(inv.InvoiceID))
			} else {
				keptCarry = append(keptCarry, inv)
			}
		}
		if len(expiredInvoiceIDs) > 0 {
			if err := invoiceModel.BatchUpdateInvoicesToExpired(expiredInvoiceIDs); err != nil {
				return result, fmt.Errorf("expire invoices: %w", err)
			}
			result.Expired = append(result.Expired, PaidItem{
				TenantID:        key.TenantID,
				BillingConfigID: key.BillingConfigID,
				Amount:          decimal.Zero,
				PushType:        "expired_carryover",
			})
			log.Info("expired stale carryover invoices",
				zap.Uint("payingID", paying.ID),
				zap.Uint("tenantID", key.TenantID),
				zap.Uint("billingConfigID", key.BillingConfigID),
				zap.Int("expiredCount", len(expiredInvoiceIDs)))
		}

		// Phase 4 applied to both current-month and carried invoices:
		// non-commitment invoices under a managed_by_sales PM are parked in
		// SUSPENDED (status flip only — with the pool table gone there is
		// nothing to clear; the suspended rows simply drop out of
		// ListCarryoverInvoices / ListStripeEligibleInvoices on the next run).
		if pmForGroup != nil && pmForGroup.ManagedBySales {
			commitmentInvoices := inMonthInvoices[:0:0]
			carryCommitment := keptCarry[:0]
			mbsSuspendIDs := []uint{}
			for _, inv := range inMonthInvoices {
				if inv.Type == entity.InvoiceType_COMMITMENT {
					commitmentInvoices = append(commitmentInvoices, inv)
				} else {
					mbsSuspendIDs = append(mbsSuspendIDs, uint(inv.InvoiceID))
				}
			}
			for _, inv := range keptCarry {
				if inv.Type == entity.InvoiceType_COMMITMENT {
					carryCommitment = append(carryCommitment, inv)
				} else {
					mbsSuspendIDs = append(mbsSuspendIDs, uint(inv.InvoiceID))
				}
			}
			if len(mbsSuspendIDs) > 0 {
				if err := invoiceModel.BatchUpdateInvoicesToSuspended(mbsSuspendIDs); err != nil {
					return result, fmt.Errorf("suspend managed-by-sales non-commitment invoices: %w", err)
				}
				suspendedAmount := decimal.Zero
				for _, inv := range inMonthInvoices {
					if inv.Type != entity.InvoiceType_COMMITMENT {
						suspendedAmount = suspendedAmount.Add(inv.CostPrice)
					}
				}
				for _, inv := range keptCarry {
					if inv.Type != entity.InvoiceType_COMMITMENT {
						suspendedAmount = suspendedAmount.Add(inv.CostPrice)
					}
				}
				result.Suspended = append(result.Suspended, PaidItem{
					TenantID:        key.TenantID,
					BillingConfigID: key.BillingConfigID,
					Amount:          suspendedAmount,
					PushType:        "managed_by_sales_non_commitment",
				})
				log.Info("managed-by-sales non-commitment invoices parked in SUSPENDED",
					zap.Uint("payingID", paying.ID),
					zap.Uint("tenantID", key.TenantID),
					zap.Uint("billingConfigID", key.BillingConfigID),
					zap.Int("suspended_count", len(mbsSuspendIDs)),
					zap.Int("commitment_count", len(commitmentInvoices)+len(carryCommitment)))
			}
			// Replace both sets with the commitment-only subsets so all
			// downstream branches (threshold / push / carry) operate only on
			// the rows we still plan to push.
			inMonthInvoices = commitmentInvoices
			keptCarry = carryCommitment
			if len(inMonthInvoices) == 0 && len(keptCarry) == 0 {
				continue
			}
		}

		// Carried invoice IDs — already below_threshold_carry at run start.
		carried := make(map[uint]struct{}, len(keptCarry))
		for _, inv := range keptCarry {
			carried[uint(inv.InvoiceID)] = struct{}{}
		}

		// Combined = carried + current-month, oldest first.
		// ListCarryoverInvoices and ListStripeEligibleInvoices both order by
		// invoice_date ASC and every carried invoice predates the current
		// month, so the concat is already ordered; sort anyway as
		// defense-in-depth.
		combined := make([]forms.Invoice, 0, len(keptCarry)+len(inMonthInvoices))
		combined = append(combined, keptCarry...)
		combined = append(combined, inMonthInvoices...)
		sort.SliceStable(combined, func(i, j int) bool {
			return combined[i].StartDate.Before(combined[j].StartDate)
		})
		// Dedup by InvoiceID. The payment_methods JOIN in
		// ListStripeEligibleInvoices can return the same invoice twice when a
		// billing_config has multiple active Stripe payment methods; a
		// duplicate would double-count in runningTotal and — worse — push the
		// same invoice twice.
		seen := make(map[uint]struct{}, len(combined))
		deduped := combined[:0]
		for _, inv := range combined {
			id := uint(inv.InvoiceID)
			if _, ok := seen[id]; ok {
				continue
			}
			seen[id] = struct{}{}
			deduped = append(deduped, inv)
		}
		combined = deduped

		log.Info("processing group",
			zap.Uint("payingID", paying.ID),
			zap.Uint("tenantID", key.TenantID),
			zap.Uint("billingConfigID", key.BillingConfigID),
			zap.Int("inMonthCount", len(inMonthInvoices)),
			zap.Int("carryCount", len(keptCarry)),
			zap.Int("expiredCount", len(expiredInvoiceIDs)),
			zap.Int("combinedCount", len(combined)))

		if config.Get().GetBool(config.StripePayingPushPerInvoice) {
			// All carried invoices accumulate into the push batch up
			// front — regardless of invoice_date order. A carried
			// invoice from earlier this month can sort AFTER a new
			// invoice; accumulating them first guarantees the merged
			// push bundles every carried invoice with the crossing
			// invoice, never pushing a carried invoice alone.
			var pushBatch []forms.Invoice
			for _, inv := range combined {
				if _, ok := carried[uint(inv.InvoiceID)]; ok {
					pushBatch = append(pushBatch, inv)
				}
			}
			runningTotal := sumCostPrice(pushBatch)
			carryIDs := []uint{}
			mergedPushed := false

			for i, inv := range combined {
				if _, ok := carried[uint(inv.InvoiceID)]; ok {
					// Already accumulated into pushBatch.
					continue
				}
				runningTotal = runningTotal.Add(inv.CostPrice)
				if !mergedPushed && runningTotal.GreaterThanOrEqual(minAmount) {
					// Crossing invoice: merged push of the waiting carried
					// batch plus this invoice; the rest go per-invoice.
					pushBatch = append(pushBatch, inv)
					pushIDs := invoiceIDs(pushBatch)
					chargeMode := paymenthandler.ChargeModeManual
					pushType := "accumulated"
					// Single-month auto rule: only a merged push that bundles
					// no carried invoices and contains exactly one current-month
					// invoice auto-charges. Any carried invoice being bundled
					// makes it an accumulated balance push — the customer
					// should pay those line items at their discretion.
					if len(keptCarry) == 0 && len(pushBatch) == 1 && sameMonth(pushBatch) {
						chargeMode = paymenthandler.ChargeModeAuto
						pushType = "single_month"
					}
					stripeInvoice, err := pushInvoiceForGroup(
						ctx, key.TenantID, key.BillingConfigID,
						pmForGroup,
						pushBatch, chargeMode,
					)
					if err != nil {
						return result, fmt.Errorf("stripe push (carryover+invoice): %w", err)
					}
					stripeInvoiceID := ""
					if stripeInvoice != nil {
						stripeInvoiceID = stripeInvoice.StripeInvoiceID
						log.Info("stripe push (carryover+invoice) returned invoice",
							zap.Uint("payingID", paying.ID),
							zap.Uint("tenantID", key.TenantID),
							zap.Uint("billingConfigID", key.BillingConfigID),
							zap.String("stripeInvoiceID", stripeInvoiceID))
					}
					if err := invoiceModel.BatchUpdateInvoicesToSENT(
						pushIDs, stripeInvoiceID, string(chargeMode), pmForGroup,
					); err != nil {
						return result, fmt.Errorf("mark merged invoices SENT: %w", err)
					}
					result.Confirmed = append(result.Confirmed, PaidItem{
						TenantID:         key.TenantID,
						BillingConfigID:  key.BillingConfigID,
						MergedInvoiceIDs: pushIDs,
						Amount:           sumCostPrice(pushBatch),
						PushType:         pushType,
						ChargeMode:       string(chargeMode),
						StripeInvoiceID:  stripeInvoiceID,
					})
					// Re-check the paying heartbeat after the long merged push.
					alive, hbErr := payingModel.Heartbeat(paying.ID)
					if hbErr != nil {
						return result, fmt.Errorf("heartbeat: %w", hbErr)
					}
					if !alive {
						return result, fmt.Errorf("%w", ErrPayingNotRunning)
					}
					mergedPushed = true

					// Remaining current-month invoices go per-invoice: pay
					// those at/above the threshold, carry the rest.
					for _, rest := range combined[i+1:] {
						if _, ok := carried[uint(rest.InvoiceID)]; ok {
							// A carried invoice after the crossing point (only
							// possible with out-of-order invoice dates) keeps
							// waiting for a future merged push.
							continue
						}
						if rest.CostPrice.LessThan(minAmount) {
							carryIDs = append(carryIDs, uint(rest.InvoiceID))
							continue
						}
						pm := &paymentsub.PaymentModel{}
						payResult, payErr := pm.Pay(
							key.TenantID, key.BillingConfigID, rest,
						)
						if payErr != nil {
							log.Error("pay invoice failed",
								zap.Uint("payingID", paying.ID),
								zap.Uint("tenantID", key.TenantID),
								zap.Uint("invoiceID", uint(rest.InvoiceID)),
								zap.Error(payErr))
							result.Skipped = append(result.Skipped, PaidItem{
								TenantID:        key.TenantID,
								BillingConfigID: key.BillingConfigID,
								Amount:          rest.CostPrice,
								PushType:        "accumulated",
							})
							continue
						}
						if err := invoiceModel.UpdateSentInvoices(payResult); err != nil {
							log.Error("update sent invoices failed",
								zap.Uint("payingID", paying.ID),
								zap.Uint("tenantID", key.TenantID),
								zap.Uint("invoiceID", uint(rest.InvoiceID)),
								zap.Error(err))
							result.Skipped = append(result.Skipped, PaidItem{
								TenantID:        key.TenantID,
								BillingConfigID: key.BillingConfigID,
								Amount:          rest.CostPrice,
								PushType:        "accumulated",
							})
							continue
						}
						if !paidInvoicesContains(payResult, uint(rest.InvoiceID)) {
							log.Error("invoice not in pay result",
								zap.Uint("payingID", paying.ID),
								zap.Uint("tenantID", key.TenantID),
								zap.Uint("invoiceID", uint(rest.InvoiceID)))
							result.Skipped = append(result.Skipped, PaidItem{
								TenantID:        key.TenantID,
								BillingConfigID: key.BillingConfigID,
								Amount:          rest.CostPrice,
								PushType:        "accumulated",
							})
							continue
						}
						result.Confirmed = append(result.Confirmed, PaidItem{
							TenantID:         key.TenantID,
							BillingConfigID:  key.BillingConfigID,
							MergedInvoiceIDs: []uint{uint(rest.InvoiceID)},
							Amount:           rest.CostPrice,
							PushType:         "accumulated",
						})
						alive, hbErr := payingModel.Heartbeat(paying.ID)
						if hbErr != nil {
							return result, fmt.Errorf("heartbeat: %w", hbErr)
						}
						if !alive {
							return result, fmt.Errorf("%w", ErrPayingNotRunning)
						}
						// Pace Stripe pushes: one per second per confirmed send,
						// giving the push rate limiter room between invoices.
						time.Sleep(time.Second)
					}
					break
				}
				if _, ok := carried[uint(inv.InvoiceID)]; ok {
					// Carried invoice: always accumulates into the push batch.
					pushBatch = append(pushBatch, inv)
					continue
				}
				// New invoice below the running threshold: flip to
				// below_threshold_carry for next month (status only — no pool
				// write). It does NOT join this run's merged push.
				carryIDs = append(carryIDs, uint(inv.InvoiceID))
			}

			// Flip every newly-carried invoice in one batch.
			if len(carryIDs) > 0 {
				if err := invoiceModel.BatchUpdateInvoicesToCarry(carryIDs); err != nil {
					return result, fmt.Errorf("carry invoices: %w", err)
				}
				byID := make(map[uint]decimal.Decimal, len(combined))
				for _, inv := range combined {
					byID[uint(inv.InvoiceID)] = inv.CostPrice
				}
				for _, id := range carryIDs {
					result.CarryoverUpdated = append(result.CarryoverUpdated, PaidItem{
						TenantID:        key.TenantID,
						BillingConfigID: key.BillingConfigID,
						Amount:          byID[id],
						PushType:        "below_threshold_carry",
					})
				}
				log.Info("carried below-threshold invoices",
					zap.Uint("payingID", paying.ID),
					zap.Uint("tenantID", key.TenantID),
					zap.Uint("billingConfigID", key.BillingConfigID),
					zap.Int("carriedCount", len(carryIDs)))
			}
		} else {
			// Merged-all mode (StripePayingPushPerInvoice=false): one Stripe
			// invoice for the whole combined set when it clears the threshold;
			// otherwise every current-month invoice is carried (status flip
			// only — with the pool table gone, no big/small split is needed:
			// when the total is below the threshold no single invoice can be
			// at or above it).
			total := sumCostPrice(combined)
			if total.GreaterThanOrEqual(minAmount) {
				chargeMode := paymenthandler.ChargeModeManual
				pushType := "accumulated"
				if len(keptCarry) == 0 && len(inMonthInvoices) == 1 && sameMonth(inMonthInvoices) {
					chargeMode = paymenthandler.ChargeModeAuto
					pushType = "single_month"
				}
				allIDs := invoiceIDs(combined)
				log.Info("pushing merged batch to Stripe",
					zap.Uint("payingID", paying.ID),
					zap.Uint("tenantID", key.TenantID),
					zap.Uint("billingConfigID", key.BillingConfigID),
					zap.Int("mergedCount", len(allIDs)),
					zap.String("pushType", pushType),
					zap.String("chargeMode", string(chargeMode)))
				stripeInvoice, err := pushInvoiceForGroup(
					ctx, key.TenantID, key.BillingConfigID,
					pmForGroup,
					combined, chargeMode,
				)
				if err != nil {
					log.Error("stripe push (merged) failed",
						zap.Uint("payingID", paying.ID),
						zap.Uint("tenantID", key.TenantID),
						zap.Uint("billingConfigID", key.BillingConfigID),
						zap.Error(err))
					return result, fmt.Errorf("stripe push: %w", err)
				}
				stripeInvoiceID := ""
				if stripeInvoice != nil {
					stripeInvoiceID = stripeInvoice.StripeInvoiceID
				}
				if err := invoiceModel.BatchUpdateInvoicesToSENT(
					allIDs, stripeInvoiceID, string(chargeMode), pmForGroup,
				); err != nil {
					return result, fmt.Errorf("mark invoices SENT: %w", err)
				}
				alive, hbErr := payingModel.Heartbeat(paying.ID)
				if hbErr != nil {
					return result, fmt.Errorf("heartbeat: %w", hbErr)
				}
				if !alive {
					return result, fmt.Errorf("%w", ErrPayingNotRunning)
				}
				result.Confirmed = append(result.Confirmed, PaidItem{
					TenantID:         key.TenantID,
					BillingConfigID:  key.BillingConfigID,
					MergedInvoiceIDs: allIDs,
					Amount:           total,
					PushType:         pushType,
					ChargeMode:       string(chargeMode),
					StripeInvoiceID:  stripeInvoiceID,
				})
			} else {
				carryIDs := invoiceIDs(inMonthInvoices)
				if len(carryIDs) > 0 {
					if err := invoiceModel.BatchUpdateInvoicesToCarry(carryIDs); err != nil {
						return result, fmt.Errorf("carry invoices: %w", err)
					}
					result.CarryoverUpdated = append(result.CarryoverUpdated, PaidItem{
						TenantID:         key.TenantID,
						BillingConfigID:  key.BillingConfigID,
						MergedInvoiceIDs: carryIDs,
						Amount:           sumCostPrice(inMonthInvoices),
						PushType:         "below_threshold_carry",
					})
				}
			}
		}
	}

	return result, nil
}

// pushInvoiceForGroup calls the Stripe handler's PushInvoice wrapper.
//
// paymentInfo is the active payment method for this (tenant, billingConfig);
// the caller fetches it once per group so the group-level managed-by-sales
// route and the per-push PM write agree on the same PM. Returns the Stripe
// invoice (containing StripeInvoiceID) so the caller persists it via
// BatchUpdateInvoicesToSENT[WithDB] together with paymentInfo. Phase 2 of
// the invoice/PM decoupling: writing payment_method_id + payment_type +
// payment_code at SENT time is what makes the /invoices filter's
// `payment_method_id NOT IN (marketplace_set)` predicate match the row
// instead of silently hiding it via SQL NULL semantics.
func pushInvoiceForGroup(
	ctx context.Context,
	tenantID, billingConfigID uint,
	paymentInfo *forms.PaymentMethod,
	invoices []forms.Invoice,
	chargeMode paymenthandler.InvoiceChargeMode,
) (*forms.Invoice, error) {
	// No local DB calls happen in this function (the Stripe handler does
	// the push via its own clients and does not accept a ctx), so there is
	// nothing to bound here.
	if paymentInfo == nil {
		return nil, fmt.Errorf("paymentInfo is required (Phase 2)")
	}
	log.Info("calling Stripe PushInvoice",
		zap.Uint("tenantID", tenantID),
		zap.Uint("billingConfigID", billingConfigID),
		zap.String("paymentType", paymentInfo.PaymentType),
		zap.Int("invoiceCount", len(invoices)),
		zap.String("chargeMode", string(chargeMode)))
	billingInfoModel := &paymentsub.BillingInformationModel{}
	tenantCompany, tcErr := billingInfoModel.GetTenantBillingCompany(tenantID)
	var billingCompany *forms.BillingCompany
	if tcErr == nil && tenantCompany != nil && tenantCompany.CompanyKey != "" {
		bc, bcErr := billingInfoModel.GetBillingCompany(tenantCompany.CompanyKey)
		if bcErr == nil && bc != nil {
			billingCompany = bc
		}
	}
	if billingCompany == nil {
		billingCompany = &forms.BillingCompany{}
	}
	handler, err := paymenthandler.GetPaymentHandler().GetProvider(paymentInfo.PaymentType)
	if err != nil {
		return nil, err
	}
	stripeHandler, ok := handler.(paymenthandler.StripePaymentHandler)
	if !ok {
		return nil, fmt.Errorf("payment provider %q is not stripe", paymentInfo.PaymentType)
	}
	return stripeHandler.PushInvoice(paymentInfo, billingCompany, invoices, chargeMode)
}

// helpers

func invoiceIDs(invoices []forms.Invoice) []uint {
	ids := make([]uint, 0, len(invoices))
	for _, inv := range invoices {
		ids = append(ids, uint(inv.InvoiceID))
	}
	return ids
}

// paidInvoicesContains checks whether invoiceID appears in the pay result.
func paidInvoicesContains(r *forms.PaymentResult, invoiceID uint) bool {
	if r == nil {
		return false
	}
	for _, inv := range r.PaidInvoices {
		if uint(inv.InvoiceID) == invoiceID {
			return true
		}
	}
	return false
}

func sumCostPrice(invoices []forms.Invoice) decimal.Decimal {
	total := decimal.Zero
	for _, inv := range invoices {
		total = total.Add(inv.CostPrice)
	}
	return total
}
func joinIDs(ids []uint) string {
	parts := make([]string, 0, len(ids))
	for _, id := range ids {
		parts = append(parts, fmt.Sprintf("%d", id))
	}
	return strings.Join(parts, ",")
}