descrybe/apps/api/internal/billing/service.go

package billing

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Service struct {
	Pool *pgxpool.Pool

	// Hot-path caches: costs are seeded once per process and rarely change.
	costsSeeded atomic.Bool
	costMu      sync.RWMutex
	costCache   map[string]int

	// Platform feature gates are global and change rarely (admin writes).
	// Short TTL + invalidate-on-write avoids a full table read on every
	// CapabilitiesForCompany / IsAllowed / CreditsOverview call.
	gatesMu       sync.RWMutex
	gatesCache    *FeatureGatesView
	gatesCachedAt time.Time
}

const featureGatesCacheTTL = 30 * time.Second

type CreditsOverview struct {
	TotalCredits        int            `json:"total_credits"`
	UsedCredits         int            `json:"used_credits"`
	Remaining           int            `json:"remaining"`
	RemainingCredits    int            `json:"remaining_credits"`
	LowCredits          bool           `json:"low_credits"`
	LowCreditsThreshold int            `json:"low_credits_threshold"`
	ProductCount        int            `json:"product_count"`
	MaxProducts         *int           `json:"max_products,omitempty"`
	AtProductLimit      bool           `json:"at_product_limit"`
	Plan                map[string]any `json:"plan,omitempty"`
	// HasActivePlan is true only when an active company_plans row exists.
	// Missing/skipped plans still use Free entitlement gates but must not look like intentional Free / Unlimited in the UI.
	HasActivePlan bool `json:"has_active_plan"`
	// Entitlements — Free has can_use_ai=false; normalize/specs/fill still allowed.
	CanUseAI    bool `json:"can_use_ai"`
	CanUseEPREL bool `json:"can_use_eprel"`
	IsFreePlan  bool `json:"is_free_plan"`
	IsPaidPlan  bool `json:"is_paid_plan"`
	// Plan feature permissions (effective = plan ∩ global). Omitted when resolution fails.
	Features         map[string]bool `json:"features,omitempty"`
	Sections         map[string]bool `json:"sections,omitempty"`
	DisabledFeatures []string        `json:"disabled_features,omitempty"`
	FeatureETag      string          `json:"feature_etag,omitempty"`
	// Per-member access overlay (settings > team). Set by the /me handler so the
	// dashboard can say "ask your administrator" rather than "upgrade your plan".
	MemberRestricted     bool     `json:"member_restricted,omitempty"`
	MemberDeniedFeatures []string `json:"member_denied_features,omitempty"`
}

func (s *Service) CreditsOverview(ctx context.Context, companyID uuid.UUID, lowThreshold int) (CreditsOverview, error) {
	if lowThreshold <= 0 {
		lowThreshold = 100
	}
	var total, used int
	err := s.Pool.QueryRow(ctx, `
		SELECT total_credits, used_credits FROM credit_balances WHERE company_id = $1`, companyID).
		Scan(&total, &used)
	if errors.Is(err, pgx.ErrNoRows) {
		_, _ = s.Pool.Exec(ctx, `INSERT INTO credit_balances (company_id) VALUES ($1) ON CONFLICT DO NOTHING`, companyID)
		total, used = 0, 0
	} else if err != nil {
		return CreditsOverview{}, err
	}

	remaining := RemainingCreditsClamped(total, used)
	out := CreditsOverview{
		TotalCredits:        total,
		UsedCredits:         used,
		Remaining:           remaining,
		RemainingCredits:    remaining,
		LowCreditsThreshold: lowThreshold,
	}

	_ = s.Pool.QueryRow(ctx, `
		SELECT count(*) FROM processed_products WHERE company_id = $1`, companyID).Scan(&out.ProductCount)

	var planName string
	var monthlyVal int
	var monthly, maxProducts *int
	var isTrial, isCustom, isLegacy bool
	var nextBilling *time.Time
	var planNotes *string
	var featuresRaw []byte
	err = s.Pool.QueryRow(ctx, `
		SELECT p.name, p.monthly_credits, p.max_products, p.is_custom, COALESCE(p.is_legacy, false),
			cp.is_trial, cp.next_billing_date, cp.notes, COALESCE(p.features, '{}'::jsonb)
		FROM company_plans cp
		JOIN plans p ON p.id = cp.plan_id
		WHERE cp.company_id = $1 AND cp.is_active = true
		ORDER BY cp.created_at DESC LIMIT 1`, companyID).
		Scan(&planName, &monthly, &maxProducts, &isCustom, &isLegacy, &isTrial, &nextBilling, &planNotes, &featuresRaw)
	if err != nil && isUndefinedColumn(err) {
		err = s.Pool.QueryRow(ctx, `
			SELECT p.name, p.monthly_credits, p.max_products, p.is_custom,
				cp.is_trial, cp.next_billing_date, cp.notes, COALESCE(p.features, '{}'::jsonb)
			FROM company_plans cp
			JOIN plans p ON p.id = cp.plan_id
			WHERE cp.company_id = $1 AND cp.is_active = true
			ORDER BY cp.created_at DESC LIMIT 1`, companyID).
			Scan(&planName, &monthly, &maxProducts, &isCustom, &isTrial, &nextBilling, &planNotes, &featuresRaw)
		if err == nil {
			isLegacy = IsLegacyPlanName(planName)
		} else if isUndefinedColumn(err) {
			err = s.Pool.QueryRow(ctx, `
				SELECT p.name, p.monthly_credits, p.max_products, p.is_custom,
					cp.is_trial, cp.next_billing_date, cp.notes
				FROM company_plans cp
				JOIN plans p ON p.id = cp.plan_id
				WHERE cp.company_id = $1 AND cp.is_active = true
				ORDER BY cp.created_at DESC LIMIT 1`, companyID).
				Scan(&planName, &monthly, &maxProducts, &isCustom, &isTrial, &nextBilling, &planNotes)
			if err == nil {
				isLegacy = IsLegacyPlanName(planName)
				featuresRaw = []byte("{}")
			}
		}
	}
	if err == nil {
		out.HasActivePlan = true
		out.MaxProducts = maxProducts
		if maxProducts != nil && *maxProducts > 0 {
			out.AtProductLimit = out.ProductCount >= *maxProducts
		}
		if monthly != nil {
			monthlyVal = *monthly
		}
		out.Plan = map[string]any{
			"name":              planName,
			"monthly_credits":   monthlyVal,
			"max_products":      maxProducts,
			"is_custom":         isCustom,
			"is_trial":          isTrial,
			"next_billing_date": nextBilling,
		}
		if status := ParseStripeStatusNote(planNotes); status != "" {
			out.Plan["subscription_status"] = status
		}
	} else {
		// Skipped/missing company_plans: gate like Free, but HasActivePlan stays false for UX recovery.
		planName = "Free"
		featuresRaw = []byte("{}")
	}
	ent := ComputeEntitlements(planName, monthlyVal, remaining, isTrial)
	out.CanUseAI = ent.CanUseAI
	out.CanUseEPREL = ent.CanUseEPREL
	out.IsFreePlan = ent.IsFreePlan
	out.IsPaidPlan = ent.IsPaidPlan
	// Free (0 monthly credits) is not "low credits" — AI is simply unavailable.
	out.LowCredits = ent.CanUseAI && remaining <= lowThreshold && remaining > 0
	// Resolve features from the plan row already loaded (avoids a second CapabilitiesForCompany
	// round-trip that re-queries company_plans + credit_balances). Still uses GetFeatureGates cache
	// and plans.features JSON overrides via decodeFeaturesJSON + ResolveEffectiveFeaturesEx.
	if gates, gerr := s.GetFeatureGates(ctx); gerr == nil {
		if overrides, oerr := decodeFeaturesJSON(featuresRaw); oerr == nil {
			feats, sections, disabled := ResolveEffectiveFeaturesEx(
				planName, isCustom, IsLegacyPlan(planName, isLegacy), overrides, gates)
			out.Features = feats
			out.Sections = sections
			out.DisabledFeatures = disabled
			out.FeatureETag = featureETag(feats)
		}
	}
	return out, nil
}

var (
	ErrInsufficientCredits  = errors.New("insufficient credits")
	ErrProductLimitExceeded = errors.New("product limit exceeded")
)

// AIBrandApplyAllowed reports whether brand-kit voice may be injected into AI prompts.
// Free (and unknown/no plan) can edit the brand kit but AI apply is gated to paid plans
// and the capability.brand_ai_apply / marketing.brand_ai_apply feature keys.
func (s *Service) AIBrandApplyAllowed(ctx context.Context, companyID uuid.UUID) bool {
	if s == nil || s.Pool == nil {
		return false
	}
	ent, err := s.EntitlementsForCompany(ctx, companyID)
	if err != nil {
		return false
	}
	if !(ent.IsPaidPlan || ent.IsTrial) {
		return false
	}
	ok, err := s.IsAllowed(ctx, companyID, "capability.brand_ai_apply")
	if err != nil || !ok {
		return false
	}
	ok, err = s.IsAllowed(ctx, companyID, "marketing.brand_ai_apply")
	if err != nil {
		return false
	}
	return ok
}

// ProcessingGateOpts controls credit checks for a processing job start.
type ProcessingGateOpts struct {
	// RequiresAI when true enforces can_use_ai + credit wallet (AI-only job types).
	RequiresAI bool
	// RequiresEPREL when true enforces can_use_eprel (EPREL-only job types).
	RequiresEPREL bool
}

// AssertCanStartProcessing enforces plan SKU caps always; credit wallet only when RequiresAI.
// Free-tier normalize/specs/fill jobs pass with 0 credits.
func (s *Service) AssertCanStartProcessing(ctx context.Context, companyID uuid.UUID, batchSize int, opts ProcessingGateOpts) error {
	if batchSize <= 0 {
		return errors.New("no products selected")
	}

	ent, err := s.EntitlementsForCompany(ctx, companyID)
	if err != nil {
		return err
	}
	if opts.RequiresEPREL && !ent.CanUseEPREL {
		return fmt.Errorf("%w — EPREL is included on every plan; check platform EPREL settings", ErrEPRELRequiresUpgrade)
	}
	if opts.RequiresAI {
		if !ent.CanUseAI {
			return fmt.Errorf("%w — upgrade your plan or add AI credits", ErrAIRequiresUpgrade)
		}
		if ent.RemainingCredits < 1 {
			return fmt.Errorf("%w: no credits remaining — upgrade your plan to continue AI processing", ErrInsufficientCredits)
		}
		if ent.RemainingCredits < batchSize {
			return fmt.Errorf("%w: need at least %d credits for this batch (have %d) — upgrade or select fewer products",
				ErrInsufficientCredits, batchSize, ent.RemainingCredits)
		}
	}

	var maxProducts *int
	err = s.Pool.QueryRow(ctx, `
		SELECT p.max_products
		FROM company_plans cp
		JOIN plans p ON p.id = cp.plan_id
		WHERE cp.company_id = $1 AND cp.is_active = true
		ORDER BY cp.created_at DESC LIMIT 1`, companyID).Scan(&maxProducts)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	if maxProducts == nil || *maxProducts <= 0 {
		return nil
	}

	var productCount int
	if err := s.Pool.QueryRow(ctx, `
		SELECT count(*) FROM processed_products WHERE company_id = $1`, companyID).Scan(&productCount); err != nil {
		return err
	}
	slotsLeft := *maxProducts - productCount
	if slotsLeft <= 0 {
		return fmt.Errorf("%w: plan allows up to %d products — upgrade to process more",
			ErrProductLimitExceeded, *maxProducts)
	}
	if batchSize > slotsLeft {
		return fmt.Errorf("%w: only %d product slots left on your plan (limit %d) — upgrade or select fewer products",
			ErrProductLimitExceeded, slotsLeft, *maxProducts)
	}
	return nil
}

// ConsumeCredits debits company credits for one processed product.
// tokenCount is LLM tokens used for this item (0 = flat product cost only).
// Debit = feature base + ceil(tokens/1000)*openai_token_k.
// Free / no-AI path (tokenCount==0 and !CanUseAI): no debit — normalize/specs/fill is free.
// The wallet UPDATE is atomic (WHERE remaining >= debit) so concurrent consumes cannot go negative.
//
// Contends on the single credit_balances row per company (row lock until commit).
// Prefer this over batching when deliverables must not persist without a successful debit
// (pipeline processOne). Use ConsumeCreditsBatch only when the caller already holds
// results in memory and can discard them all on ErrInsufficientCredits.
func (s *Service) ConsumeCredits(ctx context.Context, companyID uuid.UUID, tokenCount int, featureName string) error {
	return s.consumeCredits(ctx, nil, companyID, tokenCount, featureName, 1)
}

// ConsumeCreditsTx applies the same debit as ConsumeCredits on an existing
// transaction so callers (processOne) can commit debit+persist atomically.
func (s *Service) ConsumeCreditsTx(ctx context.Context, tx pgx.Tx, companyID uuid.UUID, tokenCount int, featureName string) error {
	if tx == nil {
		return fmt.Errorf("ConsumeCreditsTx: nil tx")
	}
	return s.consumeCredits(ctx, tx, companyID, tokenCount, featureName, 1)
}

// ConsumeCreditsBatch applies one wallet/cycle update for productCount items and
// combined tokenCount. Reduces credit_balances lock acquisitions vs N×ConsumeCredits.
// Debit uses DebitAmountN (packs on combined tokens) — may undercharge vs summing
// per-item DebitAmount when token packs cross 1k boundaries; for exact parity sum
// DebitAmount per item and prefer N×ConsumeCredits (or a future debit-amount API).
func (s *Service) ConsumeCreditsBatch(ctx context.Context, companyID uuid.UUID, tokenCount, productCount int, featureName string) error {
	if productCount < 1 {
		productCount = 1
	}
	return s.consumeCredits(ctx, nil, companyID, tokenCount, featureName, productCount)
}

func (s *Service) consumeCredits(ctx context.Context, tx pgx.Tx, companyID uuid.UUID, tokenCount int, featureName string, productCount int) error {
	if featureName == "" {
		featureName = "product_processing"
	}
	if tokenCount < 0 {
		tokenCount = 0
	}
	if productCount < 1 {
		productCount = 1
	}
	// Entitlements gate only matters for flat (0-token) debits on Free — skip the
	// two-query load on the AI hot path where tokenCount > 0.
	if tokenCount == 0 {
		ent, err := s.EntitlementsForCompany(ctx, companyID)
		if err == nil && !ent.CanUseAI {
			return nil
		}
	}
	_ = s.EnsureDefaultCosts(ctx)

	featureCost := s.lookupCost(ctx, featureName, 1)
	tokenKCost := 1
	if tokenCount > 0 {
		tokenKCost = s.lookupCost(ctx, "openai_token_k", 1)
	}
	var debit int
	if productCount == 1 {
		debit = DebitAmount(featureCost, tokenKCost, tokenCount)
	} else {
		debit = DebitAmountN(featureCost, tokenKCost, tokenCount, productCount)
	}
	return s.applyCreditDebit(ctx, tx, companyID, debit, productCount)
}

// applyCreditDebit atomically increments used_credits and open-cycle usage.
// Ensure-row + one CTE (balance + cycle) keeps the credit_balances row lock for two
// round-trips instead of an out-of-TX insert plus two separate UPDATEs.
// When tx is nil, begins and commits its own transaction; otherwise runs on tx
// without committing (caller owns the transaction).
func (s *Service) applyCreditDebit(ctx context.Context, tx pgx.Tx, companyID uuid.UUID, debit, productCount int) error {
	if debit < 1 {
		debit = 1
	}
	if productCount < 1 {
		productCount = 1
	}

	ownTx := tx == nil
	if ownTx {
		begun, err := s.Pool.Begin(ctx)
		if err != nil {
			return err
		}
		defer begun.Rollback(ctx)
		tx = begun
	}

	// Separate from the debit CTE: Postgres data-modifying CTEs share one snapshot
	// and cannot see sibling INSERT effects in the same statement.
	_, err := tx.Exec(ctx, `
		INSERT INTO credit_balances (company_id) VALUES ($1)
		ON CONFLICT (company_id) DO NOTHING`, companyID)
	if err != nil {
		return err
	}

	var balRows, cycRows int
	err = tx.QueryRow(ctx, `
		WITH upd AS (
			UPDATE credit_balances
			SET used_credits = used_credits + $2, updated_at = now()
			WHERE company_id = $1 AND (total_credits - used_credits) >= $2
			RETURNING company_id
		),
		cyc AS (
			UPDATE billing_cycles
			SET credits_used = credits_used + $2,
				products_processed = products_processed + $3,
				updated_at = now()
			WHERE id = (
				SELECT id FROM billing_cycles
				WHERE company_id = $1 AND end_date > now()
				ORDER BY start_date DESC LIMIT 1
			)
			AND EXISTS (SELECT 1 FROM upd)
			RETURNING id
		)
		SELECT
			(SELECT count(*)::int FROM upd),
			(SELECT count(*)::int FROM cyc)`, companyID, debit, productCount).
		Scan(&balRows, &cycRows)
	if err != nil {
		return err
	}
	if balRows == 0 {
		return ErrInsufficientCredits
	}
	if cycRows == 0 {
		if err := s.ensureOpenBillingCycle(ctx, tx, companyID, debit, productCount); err != nil {
			return err
		}
	}
	if ownTx {
		return tx.Commit(ctx)
	}
	return nil
}

// ensureOpenBillingCycle opens a cycle from the active company_plan window when missing.
// When poolTx is nil, uses the service pool. initialUsed/products seed the new row (usually the debit just applied).
func (s *Service) ensureOpenBillingCycle(ctx context.Context, tx pgx.Tx, companyID uuid.UUID, initialUsed, products int) error {
	exec := s.Pool.Exec
	queryRow := s.Pool.QueryRow
	if tx != nil {
		exec = tx.Exec
		queryRow = tx.QueryRow
	}
	var start, end time.Time
	err := queryRow(ctx, `
		SELECT billing_cycle_start, next_billing_date
		FROM company_plans
		WHERE company_id = $1 AND is_active = true
		ORDER BY created_at DESC LIMIT 1`, companyID).Scan(&start, &end)
	if err != nil {
		now := time.Now().UTC()
		start, end = now, now.AddDate(0, 1, 0)
	}
	_, err = exec(ctx, `
		INSERT INTO billing_cycles (company_id, start_date, end_date, credits_used, products_processed)
		VALUES ($1, $2, $3, $4, $5)`, companyID, start, end, initialUsed, products)
	return err
}

func (s *Service) lookupCost(ctx context.Context, feature string, fallback int) int {
	s.costMu.RLock()
	if s.costCache != nil {
		if cost, ok := s.costCache[feature]; ok {
			s.costMu.RUnlock()
			if cost <= 0 {
				return fallback
			}
			return cost
		}
	}
	s.costMu.RUnlock()

	var cost int
	err := s.Pool.QueryRow(ctx, `
		SELECT cost_per_unit FROM processing_costs
		WHERE feature_name = $1 AND is_active = true`, feature).Scan(&cost)
	if err != nil || cost <= 0 {
		return fallback
	}

	s.costMu.Lock()
	if s.costCache == nil {
		s.costCache = make(map[string]int, 8)
	}
	s.costCache[feature] = cost
	s.costMu.Unlock()
	return cost
}

// EnsureDefaultCosts seeds plan-aligned processing cost rows (idempotent).
// After a successful seed in this process, subsequent calls no-op (worker seeds at boot).
func (s *Service) EnsureDefaultCosts(ctx context.Context) error {
	if s.costsSeeded.Load() {
		return nil
	}
	_, err := s.Pool.Exec(ctx, `
		INSERT INTO processing_costs (feature_name, cost_per_unit, description, is_active)
		VALUES
			('product_processing', 1, 'Credits per processed product (base)', true),
			('openai_token_k', 1, 'Credits per 1000 LLM tokens', true),
			('seo_meta_ai', 1, 'Credits per SEO AI meta apply (base)', true),
			('campaign_copy', 1, 'Credits per campaign AI generate (base)', true)
		ON CONFLICT (feature_name) DO NOTHING`)
	if err == nil {
		s.costsSeeded.Store(true)
	}
	return err
}

// EstimateDebit returns the credit cost for a feature + token pack (same math as ConsumeCredits).
func (s *Service) EstimateDebit(ctx context.Context, featureName string, tokenCount int) int {
	if featureName == "" {
		featureName = "product_processing"
	}
	_ = s.EnsureDefaultCosts(ctx)
	featureCost := s.lookupCost(ctx, featureName, 1)
	tokenKCost := 1
	if tokenCount > 0 {
		tokenKCost = s.lookupCost(ctx, "openai_token_k", 1)
	}
	return DebitAmount(featureCost, tokenKCost, tokenCount)
}

type Plan struct {
	ID             int64           `json:"id"`
	Name           string          `json:"name"`
	Description    *string         `json:"description,omitempty"`
	MonthlyCredits int             `json:"monthly_credits"`
	YearlyCredits  *int            `json:"yearly_credits,omitempty"`
	MaxProducts    *int            `json:"max_products,omitempty"`
	IsCustom       bool            `json:"is_custom"`
	IsLegacy       bool            `json:"is_legacy,omitempty"`
	Term           string          `json:"term"`
	Features       map[string]bool `json:"features,omitempty"`
	// ResolvedFeatures is plan_allows only (globals ignored); populated on admin list/get.
	ResolvedFeatures map[string]bool `json:"resolved_features,omitempty"`
	// AiCoverPercent is derived from PlanAICoverPercent (not a DB column).
	AiCoverPercent int `json:"ai_cover_percent,omitempty"`
}

// EnterpriseUnlimitedCredits is the managed AI credit pack for the public Enterprise plan.
// Marketing copy says "Unlimited"; the wallet still uses a finite high grant so debit accounting works.
// Demo seed assigns this plan to Platform Demo.
const EnterpriseUnlimitedCredits = 1_000_000

// defaultPublicPlans is the self-serve ladder (PRICING doc packaging).
// Free intentionally grants 0 AI credits — normalize/specs/fill only until upgrade.
// Client-specific deals (A1, Merkur trial, etc.) must never appear here or on public pricing.
func defaultPublicPlans() []Plan {
	freeDesc := "Forever free — map a sample feed, normalize & fill specs (no AI credits)"
	starterDesc := "Up to 100 SKUs, entry AI (~100 credits / ~50% cover), full WooCommerce sync"
	plusDesc := "Up to 400 SKUs, Woo+Shopify, AI (~400 credits / ~50% of credit base)"
	growthDesc := "Up to 1,200 SKUs, full stores, BYOK, AI (~1,200 credits / ~50% of credit base)"
	businessDesc := "Up to 4,000 SKUs, BYOK, AI (~4,000 credits / ~50% of credit base)"
	scaleDesc := "Up to 12,000 SKUs, AI (~12,000 credits / ~50% cover); packs/BYOK for more"
	enterpriseDesc := "Unlimited SKUs & feeds — large managed AI grant or BYOK, SLA, account team"
	return []Plan{
		{Name: "Free", Description: &freeDesc, MonthlyCredits: 0, MaxProducts: PlanMaxProducts("Free"), Term: "monthly"},
		{Name: "Starter", Description: &starterDesc, MonthlyCredits: MonthlyCreditsForPlan("Starter", 0), MaxProducts: PlanMaxProducts("Starter"), Term: "monthly"},
		{Name: "Plus", Description: &plusDesc, MonthlyCredits: MonthlyCreditsForPlan("Plus", 0), MaxProducts: PlanMaxProducts("Plus"), Term: "monthly"},
		{Name: "Growth", Description: &growthDesc, MonthlyCredits: MonthlyCreditsForPlan("Growth", 0), MaxProducts: PlanMaxProducts("Growth"), Term: "monthly"},
		{Name: "Business", Description: &businessDesc, MonthlyCredits: MonthlyCreditsForPlan("Business", 0), MaxProducts: PlanMaxProducts("Business"), Term: "monthly"},
		{Name: "Scale", Description: &scaleDesc, MonthlyCredits: MonthlyCreditsForPlan("Scale", 0), MaxProducts: PlanMaxProducts("Scale"), Term: "monthly"},
		// MaxProducts nil = unlimited SKU cap in AssertCanStartProcessing.
		{Name: "Enterprise", Description: &enterpriseDesc, MonthlyCredits: EnterpriseUnlimitedCredits, MaxProducts: PlanMaxProducts("Enterprise"), IsCustom: true, Term: "monthly"},
	}
}

// IsPublicProductPlan reports whether name is on the public marketing ladder.
// Client deals (A1, Merkur trial, legacy Basic/Professional, …) return false.
func IsPublicProductPlan(name string) bool {
	switch strings.ToLower(strings.TrimSpace(name)) {
	case "free", "starter", "plus", "growth", "business", "scale", "enterprise":
		return true
	default:
		return false
	}
}

// EnsureDefaultPlans upserts Free / Starter / Plus / Growth / Business / Scale / Enterprise by name.
// Aligns with PRICING packaging; Free monthly_credits = 0 (no AI grant on signup).
func (s *Service) EnsureDefaultPlans(ctx context.Context) error {
	for _, p := range defaultPublicPlans() {
		var id int64
		err := s.Pool.QueryRow(ctx, `
			SELECT id FROM plans WHERE lower(name) = lower($1) ORDER BY id LIMIT 1`, p.Name).Scan(&id)
		if errors.Is(err, pgx.ErrNoRows) {
			_, err = s.Pool.Exec(ctx, `
				INSERT INTO plans (name, description, monthly_credits, yearly_credits, max_products, is_custom, term)
				VALUES ($1, $2, $3, NULL, $4, $5, $6)`,
				p.Name, p.Description, p.MonthlyCredits, p.MaxProducts, p.IsCustom, p.Term)
			if err != nil {
				return err
			}
			continue
		}
		if err != nil {
			return err
		}
		// Sync public ladder meters only — never clobber plans.features overrides.
		// Named client deals (A1, Merkur, …) are not in defaultPublicPlans and stay untouched.
		_, err = s.Pool.Exec(ctx, `
			UPDATE plans SET description = $2, monthly_credits = $3, max_products = $4,
				is_custom = $5, term = $6, updated_at = now()
			WHERE id = $1`, id, p.Description, p.MonthlyCredits, p.MaxProducts, p.IsCustom, p.Term)
		if err != nil {
			return err
		}
	}
	// Global section gates default ON; empty plan.features stay unset (DefaultPlanFeatures).
	// Then Legacy plan row + A1 cohort assignment + sparse legacy feature backfill.
	if err := s.EnsureDefaultFeatureSeeds(ctx); err != nil {
		return err
	}
	if err := s.EnsureSuperAdminPlan(ctx); err != nil {
		return err
	}
	if err := s.EnsurePlanCatalogHygiene(ctx); err != nil {
		return err
	}
	return s.EnsureLegacyDefaults(ctx)
}

// ProvisionFreePlan assigns the Free plan (0 AI credits) to a new company.
// Best-effort: registration must not fail if Free is missing.
// Never falls back to another plan (Enterprise may have a lower id after seed-demo).
func (s *Service) ProvisionFreePlan(ctx context.Context, companyID uuid.UUID) error {
	if err := s.EnsureDefaultPlans(ctx); err != nil {
		return err
	}
	var planID int64
	err := s.Pool.QueryRow(ctx, `
		SELECT id FROM plans WHERE lower(name) = 'free' ORDER BY id LIMIT 1`).Scan(&planID)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	return s.AssignPlan(ctx, companyID, planID, false, 0)
}

func (s *Service) ListPlans(ctx context.Context) ([]Plan, error) {
	rows, err := s.Pool.Query(ctx, `
		SELECT id, name, description, monthly_credits, yearly_credits, max_products, is_custom, term,
			COALESCE(features, '{}'::jsonb)
		FROM plans ORDER BY id`)
	if err != nil {
		if isUndefinedColumn(err) {
			return s.listPlansWithoutFeatures(ctx)
		}
		return nil, err
	}
	defer rows.Close()
	out := make([]Plan, 0)
	for rows.Next() {
		var p Plan
		var raw []byte
		if err := rows.Scan(&p.ID, &p.Name, &p.Description, &p.MonthlyCredits, &p.YearlyCredits, &p.MaxProducts, &p.IsCustom, &p.Term, &raw); err != nil {
			return nil, err
		}
		overrides, derr := decodeFeaturesJSON(raw)
		if derr != nil {
			return nil, derr
		}
		p.Features = overrides
		p.IsLegacy = IsLegacyPlanName(p.Name)
		p.ResolvedFeatures = planFeaturesView(p.ID, p.Name, p.IsCustom, p.IsLegacy, overrides).ResolvedFeatures
		out = append(out, p)
	}
	return out, rows.Err()
}

func (s *Service) listPlansWithoutFeatures(ctx context.Context) ([]Plan, error) {
	rows, err := s.Pool.Query(ctx, `
		SELECT id, name, description, monthly_credits, yearly_credits, max_products, is_custom, term
		FROM plans ORDER BY id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]Plan, 0)
	for rows.Next() {
		var p Plan
		if err := rows.Scan(&p.ID, &p.Name, &p.Description, &p.MonthlyCredits, &p.YearlyCredits, &p.MaxProducts, &p.IsCustom, &p.Term); err != nil {
			return nil, err
		}
		p.Features = map[string]bool{}
		p.IsLegacy = IsLegacyPlanName(p.Name)
		p.ResolvedFeatures = planFeaturesView(p.ID, p.Name, p.IsCustom, p.IsLegacy, nil).ResolvedFeatures
		out = append(out, p)
	}
	return out, rows.Err()
}

// ListPublicPlans returns only Free / Starter / Plus / Growth / Business / Scale / Enterprise.
// Keeps client-specific plans (A1, Merkur trial, …) assignable via admin ListPlans.
func (s *Service) ListPublicPlans(ctx context.Context) ([]Plan, error) {
	all, err := s.ListPlans(ctx)
	if err != nil {
		return nil, err
	}
	order := map[string]int{
		"free": 0, "starter": 1, "plus": 2, "growth": 3, "business": 4, "scale": 5, "enterprise": 6,
	}
	out := make([]Plan, 0, 7)
	for _, p := range all {
		if !IsPublicProductPlan(p.Name) {
			continue
		}
		p.AiCoverPercent = PlanAICoverPercent(p.Name)
		out = append(out, p)
	}
	sort.SliceStable(out, func(i, j int) bool {
		return order[strings.ToLower(out[i].Name)] < order[strings.ToLower(out[j].Name)]
	})
	return out, nil
}

func (s *Service) UpsertPlan(ctx context.Context, p Plan) (Plan, error) {
	p.Name = strings.TrimSpace(p.Name)
	if p.Name == "" {
		return Plan{}, ErrPlanNameRequired
	}
	if p.Term == "" {
		p.Term = "monthly"
	}
	creating := p.ID == 0
	featuresProvided := p.Features != nil
	prepareCustomPackageCreateFeatures(&p, creating, featuresProvided)
	if IsLegacyPlan(p.Name, p.IsLegacy) {
		p.IsLegacy = true
	}
	if p.Features != nil {
		if err := validateFeatureOverrides(p.Features); err != nil {
			return Plan{}, err
		}
	}
	featuresJSON, err := encodeFeaturesJSON(p.Features)
	if err != nil {
		return Plan{}, err
	}
	if p.ID > 0 {
		if p.Features != nil {
			_, err = s.Pool.Exec(ctx, `
				UPDATE plans SET name=$2, description=$3, monthly_credits=$4, yearly_credits=$5,
					max_products=$6, is_custom=$7, term=$8, features=$9::jsonb, is_legacy=$10, updated_at=now()
				WHERE id=$1`, p.ID, p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term, featuresJSON, p.IsLegacy)
		} else {
			_, err = s.Pool.Exec(ctx, `
				UPDATE plans SET name=$2, description=$3, monthly_credits=$4, yearly_credits=$5,
					max_products=$6, is_custom=$7, term=$8, is_legacy=$9, updated_at=now()
				WHERE id=$1`, p.ID, p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term, p.IsLegacy)
		}
		if err != nil {
			if isUndefinedColumn(err) {
				// Pre-is_legacy migration: fall back without the column.
				if p.Features != nil {
					_, err = s.Pool.Exec(ctx, `
						UPDATE plans SET name=$2, description=$3, monthly_credits=$4, yearly_credits=$5,
							max_products=$6, is_custom=$7, term=$8, features=$9::jsonb, updated_at=now()
						WHERE id=$1`, p.ID, p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term, featuresJSON)
				} else {
					_, err = s.Pool.Exec(ctx, `
						UPDATE plans SET name=$2, description=$3, monthly_credits=$4, yearly_credits=$5,
							max_products=$6, is_custom=$7, term=$8, updated_at=now()
						WHERE id=$1`, p.ID, p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term)
				}
				if err != nil && p.Features != nil && isUndefinedColumn(err) {
					return Plan{}, errors.New("plans.features column missing — run migration 026_plan_features")
				}
			}
			if err != nil {
				return Plan{}, err
			}
		}
	} else {
		if p.Features != nil {
			err = s.Pool.QueryRow(ctx, `
				INSERT INTO plans (name, description, monthly_credits, yearly_credits, max_products, is_custom, term, features, is_legacy)
				VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb,$9) RETURNING id`,
				p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term, featuresJSON, p.IsLegacy,
			).Scan(&p.ID)
		} else {
			err = s.Pool.QueryRow(ctx, `
				INSERT INTO plans (name, description, monthly_credits, yearly_credits, max_products, is_custom, term, is_legacy)
				VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING id`,
				p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term, p.IsLegacy,
			).Scan(&p.ID)
		}
		if err != nil {
			if isUndefinedColumn(err) {
				if p.Features != nil {
					err = s.Pool.QueryRow(ctx, `
						INSERT INTO plans (name, description, monthly_credits, yearly_credits, max_products, is_custom, term, features)
						VALUES ($1,$2,$3,$4,$5,$6,$7,$8::jsonb) RETURNING id`,
						p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term, featuresJSON,
					).Scan(&p.ID)
				} else {
					err = s.Pool.QueryRow(ctx, `
						INSERT INTO plans (name, description, monthly_credits, yearly_credits, max_products, is_custom, term)
						VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id`,
						p.Name, p.Description, p.MonthlyCredits, p.YearlyCredits, p.MaxProducts, p.IsCustom, p.Term,
					).Scan(&p.ID)
				}
				if err != nil && p.Features != nil && isUndefinedColumn(err) {
					return Plan{}, errors.New("plans.features column missing — run migration 026_plan_features")
				}
			}
			if err != nil {
				return Plan{}, err
			}
		}
	}
	view, gerr := s.GetPlanFeatures(ctx, p.ID)
	if gerr == nil {
		p.Features = view.Features
		p.ResolvedFeatures = view.ResolvedFeatures
		p.IsCustom = view.IsCustom
		p.IsLegacy = view.IsLegacy
		p.Name = view.PlanName
	}
	return p, nil
}

func (s *Service) AssignPlan(ctx context.Context, companyID uuid.UUID, planID int64, isTrial bool, trialCredits int) error {
	var monthly int
	err := s.Pool.QueryRow(ctx, `SELECT monthly_credits FROM plans WHERE id = $1`, planID).Scan(&monthly)
	if errors.Is(err, pgx.ErrNoRows) {
		return ErrPlanNotFound
	}
	if err != nil {
		return err
	}
	tx, err := s.Pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)

	_, err = tx.Exec(ctx, `UPDATE company_plans SET is_active = false, updated_at = now() WHERE company_id = $1 AND is_active = true`, companyID)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	next := now.AddDate(0, 1, 0)
	_, err = tx.Exec(ctx, `
		INSERT INTO company_plans (company_id, plan_id, is_active, billing_cycle_start, next_billing_date, is_trial, trial_credits)
		VALUES ($1,$2,true,$3,$4,$5,$6)`, companyID, planID, now, next, isTrial, trialCredits)
	if err != nil {
		return err
	}
	alloc := monthly
	if isTrial && trialCredits > 0 {
		alloc = trialCredits
	}
	_, err = tx.Exec(ctx, `
		INSERT INTO credit_balances (company_id, total_credits, used_credits, updated_at)
		VALUES ($1, $2, 0, now())
		ON CONFLICT (company_id) DO UPDATE SET total_credits = EXCLUDED.total_credits, used_credits = 0, updated_at = now()`,
		companyID, alloc)
	if err != nil {
		return err
	}
	// Close any still-open cycles so UsageSummary / ConsumeCredits never read stale ended rows as "current".
	_, err = tx.Exec(ctx, `
		UPDATE billing_cycles SET end_date = $2, updated_at = now()
		WHERE company_id = $1 AND end_date > $2`, companyID, now)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `
		INSERT INTO billing_cycles (company_id, start_date, end_date, credits_used, products_processed)
		VALUES ($1, $2, $3, 0, 0)`, companyID, now, next)
	if err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *Service) AddCredits(ctx context.Context, companyID uuid.UUID, amount int) error {
	if amount == 0 {
		return ErrAmountRequired
	}
	_, _ = s.Pool.Exec(ctx, `INSERT INTO credit_balances (company_id) VALUES ($1) ON CONFLICT DO NOTHING`, companyID)
	// Clawbacks (negative amount) must not push total below used_credits (no negative remaining).
	_, err := s.Pool.Exec(ctx, `
		UPDATE credit_balances
		SET total_credits = GREATEST(used_credits, GREATEST(0, total_credits + $2)), updated_at = now()
		WHERE company_id = $1`, companyID, amount)
	return err
}

// UsageDayPoint is one UTC day of company product/token activity.
type UsageDayPoint struct {
	Date     string `json:"date"`
	Products int64  `json:"products"`
	Tokens   int64  `json:"tokens"`
}

// UsageSummary is company usage for the billing UI.
// Credits always come from live credit_balances (not stale billing_cycles rows).
// Products/tokens respect Range; cycle dates come from the active company_plans row.
type UsageSummary struct {
	CompanyID         uuid.UUID       `json:"company_id"`
	Range             string          `json:"range"`
	CreditsUsed       int             `json:"credits_used"`
	CreditsTotal      int             `json:"credits_total"`
	CreditsRemaining  int             `json:"credits_remaining"`
	ProductsProcessed int             `json:"products_processed"`
	ProductsTotal     int             `json:"products_total"`
	Tokens            int64           `json:"tokens"`
	FeedsInput        int             `json:"feeds_input"`
	FeedsExport       int             `json:"feeds_export"`
	JobsTotal         int             `json:"jobs_total"`
	CycleStart        *time.Time      `json:"cycle_start,omitempty"`
	CycleEnd          *time.Time      `json:"cycle_end,omitempty"`
	Series            []UsageDayPoint `json:"series,omitempty"`
	Notes             []string        `json:"notes,omitempty"`
}

// ParseUsageRange normalizes billing usage range query values.
func ParseUsageRange(raw string) string {
	switch strings.ToLower(strings.TrimSpace(raw)) {
	case "7d", "30d", "cycle", "all":
		return strings.ToLower(strings.TrimSpace(raw))
	default:
		return "30d"
	}
}

func (s *Service) UsageSummary(ctx context.Context, companyID uuid.UUID, rangeRaw string) (UsageSummary, error) {
	rangeKey := ParseUsageRange(rangeRaw)
	out := UsageSummary{
		CompanyID: companyID,
		Range:     rangeKey,
		Notes:     make([]string, 0, 3),
	}

	var total, used int
	err := s.Pool.QueryRow(ctx, `
		SELECT total_credits, used_credits FROM credit_balances WHERE company_id = $1`, companyID).
		Scan(&total, &used)
	if errors.Is(err, pgx.ErrNoRows) {
		total, used = 0, 0
	} else if err != nil {
		return out, err
	}
	out.CreditsTotal = total
	out.CreditsUsed = used
	out.CreditsRemaining = RemainingCreditsClamped(total, used)
	out.Notes = append(out.Notes,
		"Credits are the live wallet (credit_balances). Daily credit history is not ledgered yet — range filters apply to products and tokens only.")

	var cycleStart, cycleEnd *time.Time
	_ = s.Pool.QueryRow(ctx, `
		SELECT cp.billing_cycle_start, cp.next_billing_date
		FROM company_plans cp
		WHERE cp.company_id = $1 AND cp.is_active = true
		ORDER BY cp.created_at DESC LIMIT 1`, companyID).Scan(&cycleStart, &cycleEnd)
	out.CycleStart = cycleStart
	out.CycleEnd = cycleEnd

	now := time.Now().UTC()
	var since *time.Time
	var until *time.Time
	switch rangeKey {
	case "7d":
		t := now.Truncate(24*time.Hour).AddDate(0, 0, -6)
		since = &t
	case "30d":
		t := now.Truncate(24*time.Hour).AddDate(0, 0, -29)
		since = &t
	case "cycle":
		if cycleStart != nil {
			since = cycleStart
		}
		if cycleEnd != nil {
			until = cycleEnd
		}
		if since == nil {
			out.Notes = append(out.Notes, "No active billing cycle on company_plans — showing all-time products/tokens.")
			rangeKey = "all"
			out.Range = "all"
		}
	case "all":
		// no time filter
	}

	_ = s.Pool.QueryRow(ctx, `
		SELECT COUNT(*)::int FROM input_feeds WHERE company_id = $1`, companyID).Scan(&out.FeedsInput)
	_ = s.Pool.QueryRow(ctx, `
		SELECT COUNT(*)::int FROM export_feeds WHERE company_id = $1`, companyID).Scan(&out.FeedsExport)
	_ = s.Pool.QueryRow(ctx, `
		SELECT COUNT(*)::int FROM processing_jobs WHERE company_id = $1`, companyID).Scan(&out.JobsTotal)

	// One scan of processed_products: all-time total plus optional range stats via FILTER.
	switch {
	case since != nil && until != nil:
		_ = s.Pool.QueryRow(ctx, `
			SELECT COUNT(*)::int,
				COUNT(*) FILTER (WHERE created_at >= $2 AND created_at < $3)::int,
				COALESCE(SUM(COALESCE(total_tokens, 0)) FILTER (WHERE created_at >= $2 AND created_at < $3), 0)::bigint
			FROM processed_products
			WHERE company_id = $1`,
			companyID, *since, *until).Scan(&out.ProductsTotal, &out.ProductsProcessed, &out.Tokens)
	case since != nil:
		_ = s.Pool.QueryRow(ctx, `
			SELECT COUNT(*)::int,
				COUNT(*) FILTER (WHERE created_at >= $2)::int,
				COALESCE(SUM(COALESCE(total_tokens, 0)) FILTER (WHERE created_at >= $2), 0)::bigint
			FROM processed_products
			WHERE company_id = $1`,
			companyID, *since).Scan(&out.ProductsTotal, &out.ProductsProcessed, &out.Tokens)
	default:
		_ = s.Pool.QueryRow(ctx, `
			SELECT COUNT(*)::int, COALESCE(SUM(COALESCE(total_tokens, 0)), 0)::bigint
			FROM processed_products
			WHERE company_id = $1`, companyID).Scan(&out.ProductsTotal, &out.Tokens)
		out.ProductsProcessed = out.ProductsTotal
	}

	if rangeKey == "7d" || rangeKey == "30d" || (rangeKey == "cycle" && since != nil) {
		seriesSince := now.Truncate(24*time.Hour).AddDate(0, 0, -29)
		days := 30
		if rangeKey == "7d" {
			seriesSince = now.Truncate(24*time.Hour).AddDate(0, 0, -6)
			days = 7
		} else if rangeKey == "cycle" && since != nil {
			seriesSince = since.UTC().Truncate(24 * time.Hour)
			end := now.UTC().Truncate(24 * time.Hour)
			if until != nil && until.Before(end) {
				end = until.UTC().Truncate(24 * time.Hour)
			}
			days = int(end.Sub(seriesSince).Hours()/24) + 1
			if days < 1 {
				days = 1
			}
			if days > 90 {
				days = 90
				seriesSince = end.AddDate(0, 0, -(days - 1))
			}
		}
		out.Series = s.usageDaySeries(ctx, companyID, seriesSince, days)
	}

	return out, nil
}

func (s *Service) usageDaySeries(ctx context.Context, companyID uuid.UUID, since time.Time, days int) []UsageDayPoint {
	byDay := map[string]UsageDayPoint{}
	rows, err := s.Pool.Query(ctx, `
		SELECT (created_at AT TIME ZONE 'UTC')::date AS d,
			COUNT(*)::bigint,
			COALESCE(SUM(COALESCE(total_tokens, 0)), 0)::bigint
		FROM processed_products
		WHERE company_id = $1 AND created_at >= $2
		GROUP BY 1
		ORDER BY 1`, companyID, since)
	if err == nil {
		for rows.Next() {
			var d time.Time
			var products, tokens int64
			if err := rows.Scan(&d, &products, &tokens); err != nil {
				break
			}
			key := d.UTC().Format("2006-01-02")
			byDay[key] = UsageDayPoint{Date: key, Products: products, Tokens: tokens}
		}
		rows.Close()
	}
	out := make([]UsageDayPoint, 0, days)
	for i := 0; i < days; i++ {
		key := since.AddDate(0, 0, i).UTC().Format("2006-01-02")
		if p, ok := byDay[key]; ok {
			out = append(out, p)
			continue
		}
		out = append(out, UsageDayPoint{Date: key})
	}
	return out
}

// DueBillingCyclesResult counts successful rolls and permanent per-row failures.
// Skipped claims (SKIP LOCKED / no longer due) are neither processed nor failed.
type DueBillingCyclesResult struct {
	Processed int `json:"processed"`
	Failed    int `json:"failed"`
}

// recordDueCycleAttempt updates counters for one claimAndRollDueCompanyPlan outcome.
// Permanent failures increment Failed and return a wrapped error for aggregation.
func recordDueCycleAttempt(res *DueBillingCyclesResult, rowID int64, ok bool, err error) error {
	if err != nil {
		res.Failed++
		return fmt.Errorf("company_plan %d: %w", rowID, err)
	}
	if ok {
		res.Processed++
	}
	return nil
}

// RunDueBillingCycles rolls due company_plans into a new cycle and refreshes monthly credits.
// Concurrent callers are safe: each due row is re-claimed with FOR UPDATE SKIP LOCKED inside
// its processing transaction (same pattern as processing.ClaimNext / shopify ClaimNextPendingJob).
// Per-row permanent failures are fail-closed (counted in Failed, aggregated into the returned
// error) without aborting the rest of the multi-company run.
func (s *Service) RunDueBillingCycles(ctx context.Context) (DueBillingCyclesResult, error) {
	var res DueBillingCyclesResult
	rows, err := s.Pool.Query(ctx, `
		SELECT cp.id
		FROM company_plans cp
		WHERE cp.is_active = true AND cp.next_billing_date <= now()
		ORDER BY cp.next_billing_date ASC, cp.id ASC`)
	if err != nil {
		return res, err
	}
	defer rows.Close()

	var ids []int64
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return res, err
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return res, err
	}

	var errs []error
	for _, rowID := range ids {
		ok, rollErr := s.claimAndRollDueCompanyPlan(ctx, rowID)
		if attemptErr := recordDueCycleAttempt(&res, rowID, ok, rollErr); attemptErr != nil {
			slog.Error("billing_cycle_roll_failed", "company_plan_id", rowID, "err", rollErr)
			errs = append(errs, attemptErr)
		}
	}
	if len(errs) > 0 {
		return res, errors.Join(errs...)
	}
	return res, nil
}

// claimAndRollDueCompanyPlan claims one company_plans row with FOR UPDATE SKIP LOCKED and
// rolls it when still due. ok=false, err=nil means another worker claimed it or it is no longer due.
// Insert/update/commit failures return ok=false with a non-nil error (fail-closed).
func (s *Service) claimAndRollDueCompanyPlan(ctx context.Context, rowID int64) (ok bool, err error) {
	tx, err := s.Pool.Begin(ctx)
	if err != nil {
		return false, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	var (
		companyID uuid.UUID
		start     time.Time
		next      time.Time
		monthly   int
	)
	// Claim still-due row; SKIP LOCKED yields no row when another worker holds the lock.
	err = tx.QueryRow(ctx, `
		SELECT cp.company_id, cp.billing_cycle_start, cp.next_billing_date, p.monthly_credits
		FROM company_plans cp
		JOIN plans p ON p.id = cp.plan_id
		WHERE cp.id = $1
			AND cp.is_active = true
			AND cp.next_billing_date <= now()
		FOR UPDATE OF cp SKIP LOCKED`, rowID).Scan(&companyID, &start, &next, &monthly)
	if errors.Is(err, pgx.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}

	var used int
	err = tx.QueryRow(ctx, `
		SELECT used_credits FROM credit_balances WHERE company_id = $1 FOR UPDATE`, companyID).Scan(&used)
	if errors.Is(err, pgx.ErrNoRows) {
		used = 0
	} else if err != nil {
		return false, err
	}
	_, err = tx.Exec(ctx, `
		INSERT INTO billing_cycles (company_id, start_date, end_date, credits_used, products_processed)
		VALUES ($1, $2, $3, $4, 0)`, companyID, start, next, used)
	if err != nil {
		return false, err
	}
	newStart := next
	newNext := next.AddDate(0, 1, 0)
	_, err = tx.Exec(ctx, `
		UPDATE company_plans SET billing_cycle_start = $2, next_billing_date = $3, updated_at = now()
		WHERE id = $1`, rowID, newStart, newNext)
	if err != nil {
		return false, err
	}
	// monthly_credits == 0 (Free / A1 PAYG): preserve wallet — no monthly grant to replace.
	// Otherwise a cycle roll would wipe topped-up or migrated credits (A1 dump / packs).
	if monthly > 0 {
		_, err = tx.Exec(ctx, `
			INSERT INTO credit_balances (company_id, total_credits, used_credits, updated_at)
			VALUES ($1, $2, 0, now())
			ON CONFLICT (company_id) DO UPDATE SET total_credits = EXCLUDED.total_credits, used_credits = 0, updated_at = now()`,
			companyID, monthly)
		if err != nil {
			return false, err
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	return true, nil
}