package payment

import (
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/utils/errorutils"
	"github.com/shopspring/decimal"
	"gorm.io/gorm"
)

// CarryoverSourceItem represents a single invoice entry inside the
// stripe_invoice_push_carryover.source_invoice_ids JSON array. It records the
// invoice id, the amount it contributed to the pool, and the month (yyyy-mm)
// the invoice was first held. Months are stored as a free-form string because
// the design only needs to detect "same month" vs "different month" on push.
type CarryoverSourceItem struct {
	InvoiceID uint            `json:"invoice_id"`
	Amount    decimal.Decimal `json:"amount"`
	Month     string          `json:"month"`
}

// CarryoverVersionConflictError signals that the optimistic-lock update on
// stripe_invoice_push_carryover lost a race after exhausting the in-method
// retry budget. The worker treats this as transient and releases the paying
// back to the pending queue so the next tick re-runs the same evaluation.
type CarryoverVersionConflictError struct {
	Op       string // "upsert" or "clear"
	Attempts int
	Cause    error
}

func (e *CarryoverVersionConflictError) Error() string {
	return fmt.Sprintf("carryover %s version conflict after %d attempts: %v", e.Op, e.Attempts, e.Cause)
}

func (e *CarryoverVersionConflictError) Unwrap() error {
	return e.Cause
}

// carryoverMaxAttempts is the in-method retry budget for optimistic-lock
// conflicts. Bumped from 3 → 5 with a small backoff so transient contention
// from concurrent worker pods does not propagate as a hard failure.
const carryoverMaxAttempts = 5

// carryoverRetryBackoff bounds the per-attempt sleep so a stuck row does not
// stall the worker tick for long.
const carryoverRetryBackoffMs = 50

// CarryoverModel owns CRUD on the stripe_invoice_push_carryover table. The
// table accumulates sub-threshold invoices per (tenant, billing_config) pair
// so they can be merged into one Stripe push once the running total crosses
// the configured threshold.
//
// NOTE (fix/stripe-paying-runner-bugs): the paying worker no longer reads or
// writes this table — billing2_invoices.status = 'below_threshold_carry' is
// the single source of truth for carryover (see InvoiceModel.ListCarryoverInvoices).
// This table survives as a derived cache for CarryoverModel.Summary() (the
// dashboard carryover_summary block), which will show stale data until a
// follow-up switches it to a billing2_invoices query.
//
// Concurrency: writes use an optimistic-lock check on the Version column so
// concurrent worker pods cannot double-count when they both try to merge the
// same source invoices.
type CarryoverModel struct {
}

// GetOrCreate returns the carryover row for (tenantID, billingConfigID),
// creating an empty one (with firstCarryMonth as the seed) if it does not
// exist yet. The caller usually calls this once at the start of an
// evaluation to obtain a row to upsert into.
func (m *CarryoverModel) GetOrCreate(tenantID, billingConfigID uint, firstCarryMonth time.Time) (*entity.StripeInvoicePushCarryover, error) {
	if tenantID == 0 || billingConfigID == 0 {
		return nil, billing_errors.NewInvalidInputError(errors.New("tenantID and billingConfigID must be positive"))
	}
	row := &entity.StripeInvoicePushCarryover{}
	err := db.GetDB().Where("tenant_id = ? AND billing_config_id = ?", tenantID, billingConfigID).First(row).Error
	if err == nil {
		return row, nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, errorutils.WrapSQLError(err, "get carryover")
	}
	// Create empty row. The unique key (tenant_id, billing_config_id) protects
	// against a duplicate create when two workers race — fall back to a re-read.
	row = &entity.StripeInvoicePushCarryover{
		TenantID:         tenantID,
		BillingConfigID:  billingConfigID,
		Amount:           decimal.Zero,
		SourceInvoiceIDs: "[]",
		FirstCarryMonth:  firstCarryMonth,
		Version:          0,
	}
	if err := db.GetDB().Create(row).Error; err != nil {
		// Lost the create race; re-read the row that the other worker wrote.
		var existing entity.StripeInvoicePushCarryover
		if rereadErr := db.GetDB().Where("tenant_id = ? AND billing_config_id = ?", tenantID, billingConfigID).First(&existing).Error; rereadErr == nil {
			return &existing, nil
		}
		return nil, errorutils.WrapSQLError(err, "create carryover")
	}
	return row, nil
}

// UpsertWithAmount overwrites the existing carryover row's `amount` and
// `source_invoice_ids` with the new values, or creates the row if none
// exists. The `amount` argument is treated as the *total pool value* for
// this (tenant, billing_config) after the merge — the worker computes it
// as sumCostPrice(current invoices) + carry.Amount and passes it in. The
// function deliberately does NOT add it to the existing amount, because
// the existing amount is already included in the caller's "total" via
// carry.Amount; an additive merge would double-count.
//
// Uses optimistic-lock versioning: every read-modify-write checks
// `version = ?` and bumps it by 1. If the row was modified by a concurrent
// worker between the read and the write, the UPDATE matches 0 rows and we
// retry with a fresh read. After exhausting the retry budget we return
// a CarryoverVersionConflictError so the worker can release the paying
// back to pending instead of marking it failed.
func (m *CarryoverModel) UpsertWithAmount(tenantID, billingConfigID uint, amount decimal.Decimal, sourceInvoiceIDs []uint, month string) error {
	if tenantID == 0 || billingConfigID == 0 {
		return billing_errors.NewInvalidInputError(errors.New("tenantID and billingConfigID must be positive"))
	}
	// Re-read inside a retry loop. The version check inside the UPDATE makes
	// the merge safe under contention.
	var lastErr error
	for attempt := 0; attempt < carryoverMaxAttempts; attempt++ {
		row, err := m.GetOrCreate(tenantID, billingConfigID, time.Now().UTC())
		if err != nil {
			return err
		}
		existing, err := decodeSourceItems(row.SourceInvoiceIDs)
		if err != nil {
			return err
		}
		// Dedup: mergeSourceItems replaces existing items with the same
		// InvoiceID instead of appending a duplicate, preserving each item's
		// original Month.
		newItems := mergeSourceItems(existing, sourceInvoiceIDs, amount, month)
		encoded, err := encodeSourceItems(newItems)
		if err != nil {
			return err
		}
		// Overwrite (not additive): the caller already folds the existing
		// row.Amount into `amount`, so the row's new total is just `amount`.
		newTotal := amount
		newFirstCarryMonth := row.FirstCarryMonth
		// Carryover seed is "the earliest month we ever held". Use whichever is
		// earlier between the existing first_carry_month and the new batch's month.
		if month != "" {
			batchMonth, parseErr := time.Parse("2006-01", month)
			if parseErr == nil {
				batchMonth = time.Date(batchMonth.Year(), batchMonth.Month(), 1, 0, 0, 0, 0, time.UTC)
				if batchMonth.Before(newFirstCarryMonth) {
					newFirstCarryMonth = batchMonth
				}
			}
		}
		tx := db.GetDB().Model(&entity.StripeInvoicePushCarryover{}).
			Where("id = ? AND version = ?", row.ID, row.Version).
			Updates(map[string]interface{}{
				"amount":             newTotal,
				"source_invoice_ids": encoded,
				"first_carry_month":  newFirstCarryMonth,
				"version":            row.Version + 1,
			})
		if tx.Error != nil {
			return errorutils.WrapSQLError(tx.Error, "upsert carryover amount")
		}
		if tx.RowsAffected == 1 {
			return nil
		}
		// Another worker updated the row between our read and our write.
		lastErr = errors.New("carryover version conflict")
		time.Sleep(time.Duration(carryoverRetryBackoffMs*(attempt+1)) * time.Millisecond)
	}
	return &CarryoverVersionConflictError{Op: "upsert", Attempts: carryoverMaxAttempts, Cause: lastErr}
}

// UpsertWithAmountWithDB is the transactional variant of UpsertWithAmount.
// Single attempt — the caller owns the tx and handles rollback on conflict.
// On an optimistic-lock conflict (a concurrent worker bumped the version
// between the read and the write) nothing is written and a
// CarryoverVersionConflictError is returned, so the caller's tx rolls back
// and the paying is released to pending for a fresh re-evaluation.
func (m *CarryoverModel) UpsertWithAmountWithDB(exec *gorm.DB, tenantID, billingConfigID uint, amount decimal.Decimal, sourceInvoiceIDs []uint, month string) error {
	if tenantID == 0 || billingConfigID == 0 {
		return billing_errors.NewInvalidInputError(errors.New("tenantID and billingConfigID must be positive"))
	}
	// GetOrCreate is not usable here: it queries db.GetDB() and would run
	// outside the caller's tx. Inline the same get-or-create on `exec` so the
	// read and the upsert share the transaction.
	row := &entity.StripeInvoicePushCarryover{}
	err := exec.Where("tenant_id = ? AND billing_config_id = ?", tenantID, billingConfigID).First(row).Error
	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		return errorutils.WrapSQLError(err, "get carryover")
	}
	if errors.Is(err, gorm.ErrRecordNotFound) {
		// Create an empty row inside the tx. The unique key
		// (tenant_id, billing_config_id) protects against a duplicate create;
		// fall back to a re-read if another worker won the race.
		row = &entity.StripeInvoicePushCarryover{
			TenantID:         tenantID,
			BillingConfigID:  billingConfigID,
			Amount:           decimal.Zero,
			SourceInvoiceIDs: "[]",
			FirstCarryMonth:  time.Now().UTC(),
			Version:          0,
		}
		if createErr := exec.Create(row).Error; createErr != nil {
			var existing entity.StripeInvoicePushCarryover
			if rereadErr := exec.Where("tenant_id = ? AND billing_config_id = ?", tenantID, billingConfigID).First(&existing).Error; rereadErr == nil {
				row = &existing
			} else {
				return errorutils.WrapSQLError(createErr, "create carryover")
			}
		}
	}
	existing, err := decodeSourceItems(row.SourceInvoiceIDs)
	if err != nil {
		return err
	}
	newItems := mergeSourceItems(existing, sourceInvoiceIDs, amount, month)
	encoded, err := encodeSourceItems(newItems)
	if err != nil {
		return err
	}
	newFirstCarryMonth := row.FirstCarryMonth
	if month != "" {
		if batchMonth, parseErr := time.Parse("2006-01", month); parseErr == nil {
			batchMonth = time.Date(batchMonth.Year(), batchMonth.Month(), 1, 0, 0, 0, 0, time.UTC)
			if batchMonth.Before(newFirstCarryMonth) {
				newFirstCarryMonth = batchMonth
			}
		}
	}
	tx := exec.Model(&entity.StripeInvoicePushCarryover{}).
		Where("id = ? AND version = ?", row.ID, row.Version).
		Updates(map[string]interface{}{
			"amount":             amount,
			"source_invoice_ids": encoded,
			"first_carry_month":  newFirstCarryMonth,
			"version":            row.Version + 1,
		})
	if tx.Error != nil {
		return tx.Error
	}
	if tx.RowsAffected == 0 {
		return &CarryoverVersionConflictError{Op: "upsert", Attempts: 1, Cause: errors.New("single-attempt version conflict")}
	}
	return nil
}

// ClearSourceInvoices removes the listed invoice IDs from the row's
// source_invoice_ids JSON array, leaving every other entry intact, and
// recomputes the `amount` column from the remaining items. If all items
// are removed, amount is reset to zero so the dashboard does not report a
// stale non-zero total for an empty pool. The per-item amounts are kept in
// sync at write time by buildSourceItems (equal split with the rounding
// remainder on the first item), so summing the kept items yields the
// correct new total without re-scanning the invoices table.
func (m *CarryoverModel) ClearSourceInvoices(tenantID, billingConfigID uint, consumedIDs []uint) error {
	return m.ClearSourceInvoicesWithDB(db.GetDB(), tenantID, billingConfigID, consumedIDs)
}

// ClearSourceInvoicesWithDB accepts an explicit *gorm.DB so the caller can
// wrap the operation in an outer transaction. The semantic guarantee is
// identical to ClearSourceInvoices; the public method is just a
// convenience wrapper that uses db.GetDB().
func (m *CarryoverModel) ClearSourceInvoicesWithDB(exec *gorm.DB, tenantID, billingConfigID uint, consumedIDs []uint) error {
	if tenantID == 0 || billingConfigID == 0 {
		return billing_errors.NewInvalidInputError(errors.New("tenantID and billingConfigID must be positive"))
	}
	if len(consumedIDs) == 0 {
		return nil
	}
	consumed := make(map[uint]struct{}, len(consumedIDs))
	for _, id := range consumedIDs {
		consumed[id] = struct{}{}
	}

	var lastErr error
	for attempt := 0; attempt < carryoverMaxAttempts; attempt++ {
		var row entity.StripeInvoicePushCarryover
		err := exec.Where("tenant_id = ? AND billing_config_id = ?", tenantID, billingConfigID).First(&row).Error
		if err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return billing_errors.NewObjectNotExistsError(err)
			}
			return errorutils.WrapSQLError(err, "load carryover")
		}
		items, err := decodeSourceItems(row.SourceInvoiceIDs)
		if err != nil {
			return err
		}
		kept := items[:0]
		for _, item := range items {
			if _, drop := consumed[item.InvoiceID]; drop {
				continue
			}
			kept = append(kept, item)
		}
		if len(kept) == len(items) {
			// Nothing to remove; avoid bumping the version unnecessarily.
			return nil
		}
		encoded, err := encodeSourceItems(kept)
		if err != nil {
			return err
		}
		// Recompute the pool total from the surviving items. buildSourceItems
		// guarantees that the per-item amounts sum back to the original total,
		// so summing `kept` is the correct new amount. Zero kept items → 0.
		newAmount := decimal.Zero
		for _, item := range kept {
			newAmount = newAmount.Add(item.Amount)
		}
		tx := exec.Model(&entity.StripeInvoicePushCarryover{}).
			Where("id = ? AND version = ?", row.ID, row.Version).
			Updates(map[string]interface{}{
				"source_invoice_ids": encoded,
				"amount":             newAmount,
				"version":            row.Version + 1,
			})
		if tx.Error != nil {
			return errorutils.WrapSQLError(tx.Error, "clear carryover source invoices")
		}
		if tx.RowsAffected == 1 {
			return nil
		}
		lastErr = errors.New("carryover version conflict")
		time.Sleep(time.Duration(carryoverRetryBackoffMs*(attempt+1)) * time.Millisecond)
	}
	return &CarryoverVersionConflictError{Op: "clear", Attempts: carryoverMaxAttempts, Cause: lastErr}
}

// ListOldestFirstCarryMonth returns the rows whose first_carry_month is the
// earliest, ordered by first_carry_month ASC. Used to populate the
// carryover_summary.oldest_first_carry_month field on the dashboard.
func (m *CarryoverModel) ListOldestFirstCarryMonth(limit int) ([]*entity.StripeInvoicePushCarryover, error) {
	if limit <= 0 {
		return nil, billing_errors.NewInvalidInputError(errors.New("limit must be positive"))
	}
	var rows []*entity.StripeInvoicePushCarryover
	err := db.GetDB().Order("first_carry_month ASC").Limit(limit).Find(&rows).Error
	if err != nil {
		return nil, errorutils.WrapSQLError(err, "list oldest carryover")
	}
	return rows, nil
}

// Count returns the number of carryover rows and the sum of their amounts.
// Used by the dashboard's carryover_summary block.
func (m *CarryoverModel) Count() (int64, decimal.Decimal, error) {
	var count int64
	if err := db.GetDB().Model(&entity.StripeInvoicePushCarryover{}).Count(&count).Error; err != nil {
		return 0, decimal.Zero, errorutils.WrapSQLError(err, "count carryover")
	}
	var total decimal.Decimal
	row := db.GetDB().Model(&entity.StripeInvoicePushCarryover{}).
		Select("COALESCE(SUM(amount), 0)").
		Row()
	if err := row.Scan(&total); err != nil {
		return count, decimal.Zero, errorutils.WrapSQLError(err, "sum carryover amount")
	}
	if !total.IsPositive() {
		total = decimal.Zero
	}
	return count, total, nil
}

// CarryoverSummary aggregates pool stats for the dashboard's
// carryover_summary block. OldestFirstCarryMonth is nil when the pool is
// empty so the API layer can render it as null.
type CarryoverSummary struct {
	TotalCount            int64
	TotalAmount           decimal.Decimal
	OldestFirstCarryMonth *time.Time
}

// Summary returns the dashboard rollup in one call: total row count, total
// amount across all rows, and the earliest first_carry_month. When the pool
// is empty, OldestFirstCarryMonth is nil (not the zero time), so the API can
// distinguish "no carryovers" from "first month is January 0001".
func (m *CarryoverModel) Summary() (*CarryoverSummary, error) {
	count, total, err := m.Count()
	if err != nil {
		return nil, err
	}
	summary := &CarryoverSummary{
		TotalCount:  count,
		TotalAmount: total,
	}
	if count == 0 {
		return summary, nil
	}
	var oldest time.Time
	err = db.GetDB().Model(&entity.StripeInvoicePushCarryover{}).
		Select("MIN(first_carry_month)").
		Scan(&oldest).Error
	if err != nil {
		return nil, errorutils.WrapSQLError(err, "min carryover first_carry_month")
	}
	if !oldest.IsZero() {
		summary.OldestFirstCarryMonth = &oldest
	}
	return summary, nil
}

// FindExpired returns rows whose first_carry_month is strictly older than
// `olderThan`. Used by the worker to detect >12-month-old carryovers and
// flip them to expired_carryover status.
func (m *CarryoverModel) FindExpired(olderThan time.Time) ([]*entity.StripeInvoicePushCarryover, error) {
	var rows []*entity.StripeInvoicePushCarryover
	err := db.GetDB().Where("first_carry_month < ?", olderThan).
		Order("first_carry_month ASC").
		Find(&rows).Error
	if err != nil {
		return nil, errorutils.WrapSQLError(err, "find expired carryover")
	}
	return rows, nil
}

// mergeSourceItems deduplicates existing items with new sourceInvoiceIDs,
// preserving the original Month of existing entries. Returns the merged slice.
func mergeSourceItems(existing []CarryoverSourceItem, sourceInvoiceIDs []uint, amount decimal.Decimal, month string) []CarryoverSourceItem {
	byID := make(map[uint]int, len(existing))
	for i, it := range existing {
		byID[it.InvoiceID] = i
	}
	newItems := make([]CarryoverSourceItem, 0, len(existing)+len(sourceInvoiceIDs))
	newItems = append(newItems, existing...)
	for _, it := range buildSourceItems(sourceInvoiceIDs, amount, month) {
		if existingIdx, ok := byID[it.InvoiceID]; ok {
			it.Month = newItems[existingIdx].Month
			newItems[existingIdx] = it
			continue
		}
		newItems = append(newItems, it)
	}
	return newItems
}

// buildSourceItems creates one CarryoverSourceItem per invoice id, distributing
// `amount` evenly across the batch. The design only persists invoice id +
// total amount, but storing per-item amounts lets ClearSourceInvoices
// re-derive the column total without re-scanning the invoices table.
func buildSourceItems(ids []uint, total decimal.Decimal, month string) []CarryoverSourceItem {
	if len(ids) == 0 {
		return nil
	}
	// Equal split with the rounding remainder placed on the first item so the
	// sum is exactly `total`.
	each := total.Div(decimal.NewFromInt(int64(len(ids))))
	out := make([]CarryoverSourceItem, 0, len(ids))
	for i, id := range ids {
		amt := each
		if i == 0 {
			amt = total.Sub(each.Mul(decimal.NewFromInt(int64(len(ids) - 1))))
		}
		out = append(out, CarryoverSourceItem{
			InvoiceID: id,
			Amount:    amt,
			Month:     month,
		})
	}
	return out
}

func encodeSourceItems(items []CarryoverSourceItem) (string, error) {
	if items == nil {
		items = []CarryoverSourceItem{}
	}
	b, err := json.Marshal(items)
	if err != nil {
		return "", errorutils.WrapSQLError(err, "encode carryover source items")
	}
	return string(b), nil
}

func decodeSourceItems(raw string) ([]CarryoverSourceItem, error) {
	if raw == "" {
		return []CarryoverSourceItem{}, nil
	}
	var items []CarryoverSourceItem
	if err := json.Unmarshal([]byte(raw), &items); err != nil {
		return nil, errorutils.WrapSQLError(err, "decode carryover source items")
	}
	return items, nil
}

// DecodeCarryoverSourceItemsForTest exposes the package-private JSON decoder
// to the test harness. Production code does not need this: the worker reads
// rows through CarryoverModel methods and merges them via the model layer.
func DecodeCarryoverSourceItemsForTest(raw string) ([]CarryoverSourceItem, error) {
	return decodeSourceItems(raw)
}