package payment

import (
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"
	"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/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/timeutils"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

// Paying task lifecycle states, stored in stripe_invoice_paying.status.
//
// pending   — created by Finance UI / POST /paying, awaiting worker pickup.
// running   — worker has claimed the task and is executing evaluation.
// completed — evaluation finished (result JSON populated).
// failed    — evaluation failed (error_message populated; result stays nil).
const (
	PayingStatusPending   = "pending"
	PayingStatusRunning   = "running"
	PayingStatusCompleted = "completed"
	PayingStatusFailed    = "failed"
)

// PayingModel owns CRUD on the stripe_invoice_paying table. The model is used by:
//   - api/paying.go handlers (POST/GET /paying) for creating and listing tasks;
//   - workers/stripe_invoice_paying_worker.go for claiming and finalising tasks.
type PayingModel struct {
}

// CreatePaying inserts a new pending paying task for the given month.
//
// When force is false, the call returns billing_errors.PayingAlreadyActiveError
// if a pending or running paying already exists for the same month (the DB
// unique key uk_active_paying enforces this — the API layer maps the error
// to HTTP 409).
//
// When force is true, the call is for the Ops emergency path: any existing
// pending/running row for the same month is hard-deleted inside the same
// transaction before the new pending row is inserted. Terminal rows
// (completed / failed) are preserved — completed ones encode the audit trail
// for the previous run, and failed ones preserve the last error message.
// force=true requires the caller to have already passed the Ops RBAC check
// (the API handler is responsible for that gate).
func (m *PayingModel) CreatePaying(month string, createdBy string, force bool) (*entity.StripeInvoicePaying, error) {
	if month == "" {
		return nil, billing_errors.NewInvalidInputError(errors.New("month is required"))
	}
	if createdBy == "" {
		return nil, billing_errors.NewInvalidInputError(errors.New("createdBy is required"))
	}
	// active_flag is a STORED GENERATED column on the DB side — the
	// application never writes it; the value is derived from status.
	// Setting it here would be a no-op because the gorm `->` tag on
	// the field blocks INSERT.
	paying := &entity.StripeInvoicePaying{
		Month:     month,
		Status:    PayingStatusPending,
		CreatedBy: createdBy,
	}

	if !force {
		err := db.GetDB().Create(paying).Error
		if err != nil {
			// uk_active_paying violation -> active paying already exists for this month.
			if isUniqueActivePayingViolation(err) {
				return nil, billing_errors.NewPayingAlreadyActiveError(err)
			}
			return nil, errorutils.WrapSQLError(err, "create stripe invoice paying")
		}
		return paying, nil
	}

	// force=true: blow away any pending/running rows for this month, then
	// insert. The whole sequence runs in one transaction so we never leave
	// the table in a state where the new row's active_flag collides with a
	// row that a concurrent worker just inserted.
	//
	// We must use Unscoped().Delete() because gorm.Model carries a soft-
	// delete column; a plain Delete() would set deleted_at but leave the
	// row physically present, so uk_active_paying would still fire on the
	// new insert.
	err := db.GetDB().Transaction(func(tx *gorm.DB) error {
		del := tx.Where("month = ? AND status IN (?, ?)", month, PayingStatusPending, PayingStatusRunning).
			Unscoped().Delete(&entity.StripeInvoicePaying{})
		if del.Error != nil {
			return errorutils.WrapSQLError(del.Error, "force-delete active paying")
		}
		if err := tx.Create(paying).Error; err != nil {
			return errorutils.WrapSQLError(err, "force-create stripe invoice paying")
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return paying, nil
}

// ClaimNextPendingPaying atomically picks the oldest pending paying and marks
// it running. Returns nil, nil when there is no pending paying. Uses a single
// UPDATE statement with ORDER BY created_at ASC LIMIT 1 so multiple worker
// pods cannot claim the same row.
//
// A claim UUID is written into the result column (as JSON) so the subsequent
// SELECT can match on that exact value — no clock-precision race from
// DATETIME(3) truncation, no ambiguity about which running row we claimed.
// CompletePaying overwrites result with the real payload; FailPaying leaves
// it in place so the claim trail is visible in the dashboard.
func (m *PayingModel) ClaimNextPendingPaying() (*entity.StripeInvoicePaying, error) {
	now := timeutils.GetNow()
	claimID := uuid.New().String()
	// Temporary JSON payload so the SELECT can match exactly on this claim.
	// The row's result column is NULL at creation; writing a claim marker
	// here is harmless because CompletePaying overwrites it with the real
	// result, and FailPaying intentionally leaves the marker visible for
	// ops to inspect.
	claimMarker := fmt.Sprintf(`{"_claim_id":"%s"}`, claimID)

	tx := db.GetDB().Exec(
		"UPDATE stripe_invoice_paying SET status = ?, picked_up_at = ?, updated_at = ?, result = ? "+
			"WHERE status = ? ORDER BY created_at ASC LIMIT 1",
		PayingStatusRunning, now, now, claimMarker, PayingStatusPending,
	)
	if tx.Error != nil {
		return nil, errorutils.WrapSQLError(tx.Error, "claim next pending paying")
	}
	if tx.RowsAffected == 0 {
		return nil, nil
	}

	// Re-read by claim marker — the only running row with this JSON payload
	// is the one we just claimed. No clock comparison needed.
	var paying entity.StripeInvoicePaying
	err := db.GetDB().
		Where("status = ? AND result LIKE ?", PayingStatusRunning, "%"+claimID+"%").
		Order("created_at ASC").
		First(&paying).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			log.Warn("claimed a paying row but claim marker was not found; row may be stuck in 'running'",
				zap.String("claimID", claimID),
				zap.Time("pickedUpAtAtClaim", now))
			return nil, nil
		}
		return nil, errorutils.WrapSQLError(err, "load claimed paying")
	}
	return &paying, nil
}

// CompletePaying marks the paying as completed and writes the result JSON.
//
// The WHERE clause is gated on id = ? AND status = 'running' so the
// transition is naturally idempotent at the row-state level: re-marking an
// already-completed (or failed) row returns ObjectNotExistsError because
// status='running' no longer matches. Only mutate the in-memory struct when
// RowsAffected == 1, otherwise the caller would see a stale State.
func (m *PayingModel) CompletePaying(paying *entity.StripeInvoicePaying, resultJSON string) error {
	if paying == nil {
		return billing_errors.NewInvalidInputError(errors.New("paying is nil"))
	}
	now := timeutils.GetNow()
	// active_flag is a STORED GENERATED column — the DB will recompute
	// it to NULL once status flips to completed. There is no
	// application write to it here.
	tx := db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ? AND status = ?", paying.ID, PayingStatusRunning).
		Updates(map[string]interface{}{
			"status":       PayingStatusCompleted,
			"completed_at": now,
			"result":       resultJSON,
			"updated_at":   now,
		})
	if tx.Error != nil {
		return errorutils.WrapSQLError(tx.Error, "complete paying")
	}
	if tx.RowsAffected == 0 {
		return billing_errors.NewObjectNotExistsError(errors.New("paying not in running state"))
	}
	paying.Status = PayingStatusCompleted
	paying.CompletedAt = &now
	paying.Result = &resultJSON
	paying.ActiveFlag = nil
	return nil
}

// FailPaying marks the paying as failed and writes the error message. Result
// stays nil so the dashboard can distinguish "failed with a known reason"
// from "completed with a structured outcome".
//
// Naturally idempotent at the row-state level — re-marking an already-failed
// row returns ObjectNotExistsError because status='running' no longer
// matches. The in-memory struct is only mutated when RowsAffected == 1.
func (m *PayingModel) FailPaying(paying *entity.StripeInvoicePaying, errMsg string) error {
	if paying == nil {
		return billing_errors.NewInvalidInputError(errors.New("paying is nil"))
	}
	now := timeutils.GetNow()
	// active_flag is a STORED GENERATED column — the DB will recompute
	// it to NULL once status flips to failed. There is no application
	// write to it here.
	tx := db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ? AND status = ?", paying.ID, PayingStatusRunning).
		Updates(map[string]interface{}{
			"status":        PayingStatusFailed,
			"completed_at":  now,
			"error_message": errMsg,
			"updated_at":    now,
		})
	if tx.Error != nil {
		return errorutils.WrapSQLError(tx.Error, "fail paying")
	}
	if tx.RowsAffected == 0 {
		return billing_errors.NewObjectNotExistsError(errors.New("paying not in running state"))
	}
	paying.Status = PayingStatusFailed
	paying.CompletedAt = &now
	paying.ErrorMessage = errMsg
	paying.ActiveFlag = nil
	return nil
}

// GetPaying fetches a single paying by primary key.
func (m *PayingModel) GetPaying(id uint) (*entity.StripeInvoicePaying, error) {
	var paying entity.StripeInvoicePaying
	err := db.GetDB().Where("id = ?", id).First(&paying).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, billing_errors.NewObjectNotExistsError(err)
		}
		return nil, errorutils.WrapSQLError(err, "get paying")
	}
	return &paying, nil
}

// ListRecentPayings returns the most recent payings, newest first by created_at
// DESC. limit must be > 0; callers cap the value.
func (m *PayingModel) ListRecentPayings(limit int) ([]*entity.StripeInvoicePaying, error) {
	if limit <= 0 {
		return nil, billing_errors.NewInvalidInputError(errors.New("limit must be positive"))
	}
	var payings []*entity.StripeInvoicePaying
	err := db.GetDB().Order("created_at DESC").Limit(limit).Find(&payings).Error
	if err != nil {
		return nil, errorutils.WrapSQLError(err, "list recent payings")
	}
	return payings, nil
}

// ListActivePayingByMonth finds a pending or running paying for the given
// month. Returns ObjectNotExistsError when none exists; the API layer maps
// that to a 201 (proceed) and a found row to 409 (already active).
func (m *PayingModel) ListActivePayingByMonth(month string) (*entity.StripeInvoicePaying, error) {
	if month == "" {
		return nil, billing_errors.NewInvalidInputError(errors.New("month is required"))
	}
	var paying entity.StripeInvoicePaying
	err := db.GetDB().Where("month = ? AND status IN (?, ?)", month, PayingStatusPending, PayingStatusRunning).
		Order("created_at ASC").First(&paying).Error
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, billing_errors.NewObjectNotExistsError(err)
		}
		return nil, errorutils.WrapSQLError(err, "list active paying by month")
	}
	return &paying, nil
}

// UpdatePayingResult overwrites the result JSON without changing the status.
// Used when the worker wants to persist incremental state, or when a re-run
// re-evaluates the same row.
func (m *PayingModel) UpdatePayingResult(id uint, resultJSON string) error {
	tx := db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ?", id).
		Update("result", resultJSON)
	if tx.Error != nil {
		return errorutils.WrapSQLError(tx.Error, "update paying result")
	}
	if tx.RowsAffected == 0 {
		return billing_errors.NewObjectNotExistsError(errors.New("paying not found"))
	}
	return nil
}

// ResetToPending returns a running paying back to the pending state so the
// next worker tick can re-claim it. Used after a transient carryover conflict
// where retrying inside the same tick would risk starving other payings or
// overrunning the polling interval. The active_flag STORED GENERATED column
// re-derives to 1 automatically when status flips back to pending.
//
// The update is guarded by the current status being `running` so a paying
// already finalised by another pod cannot be silently rewound.
func (m *PayingModel) ResetToPending(paying *entity.StripeInvoicePaying) error {
	if paying == nil {
		return billing_errors.NewInvalidInputError(errors.New("paying is nil"))
	}
	now := timeutils.GetNow()
	tx := db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ? AND status = ?", paying.ID, PayingStatusRunning).
		Updates(map[string]interface{}{
			"status":       PayingStatusPending,
			"picked_up_at": nil,
			"result":       nil,
			"updated_at":   now,
		})
	if tx.Error != nil {
		return errorutils.WrapSQLError(tx.Error, "reset paying to pending")
	}
	if tx.RowsAffected == 0 {
		return billing_errors.NewObjectNotExistsError(errors.New("paying not in running state"))
	}
	paying.Status = PayingStatusPending
	paying.PickedUpAt = nil
	paying.Result = nil
	flag := uint(1)
	paying.ActiveFlag = &flag
	return nil
}

// Heartbeat touches updated_at on a running paying so the janitor knows the
// worker is still alive. Only updates rows that are still status='running' —
// if CompletePaying or FailPaying flipped the row to a terminal state, this
// is a no-op so we never overwrite terminal timestamps.
//
// heartbeatInterval must be smaller than the janitor's stale-running
// threshold (10 minutes by default in workers/stripe_invoice_paying_janitor.go)
// or the janitor will mark a healthy long-running task as failed.
//
// The returned bool reports whether the heartbeat actually touched the row:
// true means RowsAffected == 1 (the row is still running and the worker is
// alive), false means the row has already flipped to a terminal state
// (completed/failed) or is missing. Callers should bail out without writing
// CompletePaying/FailPaying when the heartbeat reports false — at that point
// another worker has already finalised the row and a follow-up write would
// either fail the status guard or, worse, race with that terminal write.
func (m *PayingModel) Heartbeat(payingID uint) (bool, error) {
	now := timeutils.GetNow()
	tx := db.GetDB().Model(&entity.StripeInvoicePaying{}).
		Where("id = ? AND status = ?", payingID, PayingStatusRunning).
		Update("updated_at", now)
	if tx.Error != nil {
		return false, errorutils.WrapSQLError(tx.Error, "paying heartbeat")
	}
	return tx.RowsAffected == 1, nil
}

// ListStuckRunning returns payings that have been in status='running' with no
// heartbeat update for at least staleThreshold. Used by the janitor to find
// crashed workers so they can be force-failed and the next month's paying
// can re-claim them.
func (m *PayingModel) ListStuckRunning(staleThreshold time.Duration) ([]*entity.StripeInvoicePaying, error) {
	cutoff := timeutils.GetNow().Add(-staleThreshold)
	var payings []*entity.StripeInvoicePaying
	err := db.GetDB().
		Where("status = ?", PayingStatusRunning).
		Where("updated_at < ?", cutoff).
		Where("deleted_at IS NULL").
		Order("updated_at ASC").
		Limit(100).
		Find(&payings).Error
	if err != nil {
		return nil, errorutils.WrapSQLError(err, "list stuck running payings")
	}
	return payings, nil
}

// isUniqueActivePayingViolation returns true when the SQL error looks like a
// duplicate-key violation against the uk_active_paying unique index.
//
// The MySQL 8 driver + gorm v2's TranslateError rewrites "ERROR 1062 (23000):
// Duplicate entry '...' for key 'stripe_invoice_paying.uk_active_paying'" to
// "duplicated key not allowed" (no index name, no error code). So we have to
// match on three layers:
//  1. The raw index name (some drivers / SQL modes preserve it);
//  2. The MySQL error number 1062;
//  3. The gorm-translated "duplicated key not allowed" wording.
//
// Since the only unique key on this table that can be hit by a fresh insert
// is uk_active_paying (id is auto-increment and the only other index is a
// non-unique status+created_at pair), any duplicate-key error from
// Create() is by construction a same-month active collision.
func isUniqueActivePayingViolation(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	if strings.Contains(msg, "uk_active_paying") {
		return true
	}
	if strings.Contains(msg, "Error 1062") || strings.Contains(msg, "1062") {
		return true
	}
	return strings.Contains(msg, "duplicated key not allowed") ||
		strings.Contains(msg, "Duplicate entry")
}