package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
)

type historySequencePrelockRetryError struct {
	fields []string
	cause  error
}

func (e *historySequencePrelockRetryError) Error() string {
	return fmt.Sprintf("field-history sequence lock order requires retry with the complete field-history sequence lock set: %v", e.cause)
}

func (e *historySequencePrelockRetryError) Unwrap() error {
	return e.cause
}

func (e *historySequencePrelockRetryError) HistorySequencePrelockFields() []string {
	if e == nil {
		return nil
	}
	return append([]string(nil), e.fields...)
}

// ReserveHistorySequences reserves one monotonically increasing sequence for
// each stable FieldId inside the caller's active transaction. Field IDs are
// sorted and deduplicated per call, while transaction-local planning enforces
// the same deterministic order across every reservation call.
func ReserveHistorySequences(ctx context.Context, tx interface{}, fieldIDs []string) (map[string]int64, error) {
	exec, ok := tx.(Executor)
	if !ok || exec == nil {
		return nil, apperrors.NewInternalError("history sequence reservation requires an active transaction executor", nil)
	}
	if _, rootDB := tx.(*sql.DB); rootDB {
		return nil, apperrors.NewInternalError("history sequence reservation rejects the root database executor", nil)
	}
	planned := ports.PlannedHistorySequenceLocks(ctx)
	if len(planned) > 0 {
		if err := prelockHistorySequenceRows(ctx, tx, planned); err != nil {
			return nil, err
		}
	}
	// The first reservation closes planning even when the initial plan was
	// empty. Otherwise a later nested mutation could append a lower-sorting
	// field and bypass the existing high-water replay guard through the
	// prelock path. A late discovery intentionally falls back to typed replay.
	ports.MarkPlannedHistorySequenceLocksConsumed(ctx)

	ordered := ports.OrderedHistorySequenceFields(fieldIDs)
	if len(ordered) == 0 {
		return map[string]int64{}, nil
	}
	toAcquire, err := ports.PlanHistorySequenceLocks(ctx, ordered)
	if err != nil {
		if errors.Is(err, ports.ErrOutOfOrderHistorySequenceLock) {
			fields := append(ports.HeldHistorySequenceLockFields(ctx), ordered...)
			return nil, &historySequencePrelockRetryError{
				fields: ports.OrderedHistorySequenceFields(fields),
				cause:  fmt.Errorf("%w: %v", ports.ErrHistorySerializationConflict, err),
			}
		}
		return nil, apperrors.NewInternalError("plan field history sequence locks", err)
	}
	newLocks := make(map[string]struct{}, len(toAcquire))
	for _, fieldID := range toAcquire {
		newLocks[fieldID] = struct{}{}
	}

	for range ordered {
		ports.MarkHistorySerializationAttempt(ctx)
	}
	valuePlaceholders := make([]string, len(ordered))
	args := make([]interface{}, len(ordered))
	for i, fieldID := range ordered {
		valuePlaceholders[i] = "(?, 1)"
		args[i] = fieldID
	}
	_, err = exec.ExecContext(ctx, `
		INSERT INTO _System_FieldHistorySequence
			(FieldId, CurrentSequence)
		VALUES `+strings.Join(valuePlaceholders, ", ")+`
		ON DUPLICATE KEY UPDATE
			CurrentSequence = CurrentSequence + 1,
			LastModifiedDate = CURRENT_TIMESTAMP(6)`, args...)
	if err != nil {
		if isOptimisticWriteConflict(err) {
			return nil, historySerializationConflict("reserve field history sequence", err)
		}
		return nil, apperrors.NewInternalError("reserve field history sequence", err)
	}
	for _, fieldID := range ordered {
		if _, acquired := newLocks[fieldID]; acquired {
			ports.MarkHistorySequenceLockAcquired(ctx, fieldID)
		}
	}
	if len(ordered) == 1 {
		var sequence int64
		if err := exec.QueryRowContext(ctx, `
			SELECT CurrentSequence
			FROM _System_FieldHistorySequence
			WHERE FieldId = ?`, ordered[0]).Scan(&sequence); err != nil {
			if isOptimisticWriteConflict(err) {
				return nil, historySerializationConflict("read reserved field history sequence", err)
			}
			return nil, apperrors.NewInternalError("read reserved field history sequence", err)
		}
		if sequence < 1 {
			return nil, historySerializationConflict("reserved field history sequence is invalid", nil)
		}
		return map[string]int64{ordered[0]: sequence}, nil
	}

	queryPlaceholders := strings.TrimSuffix(strings.Repeat("?,", len(ordered)), ",")
	rows, err := exec.QueryContext(ctx, `
		SELECT FieldId, CurrentSequence
		FROM _System_FieldHistorySequence
		WHERE FieldId IN (`+queryPlaceholders+`)
		ORDER BY FieldId`, args...)
	if err != nil {
		if isOptimisticWriteConflict(err) {
			return nil, historySerializationConflict("read reserved field history sequence", err)
		}
		return nil, apperrors.NewInternalError("read reserved field history sequence", err)
	}
	defer rows.Close()

	reserved := make(map[string]int64, len(ordered))
	for rows.Next() {
		var fieldID string
		var sequence int64
		if err := rows.Scan(&fieldID, &sequence); err != nil {
			return nil, apperrors.NewInternalError("scan reserved field history sequence", err)
		}
		if sequence < 1 {
			return nil, historySerializationConflict("reserved field history sequence is invalid", nil)
		}
		reserved[fieldID] = sequence
	}
	if err := rows.Err(); err != nil {
		if isOptimisticWriteConflict(err) {
			return nil, historySerializationConflict("read reserved field history sequence", err)
		}
		return nil, apperrors.NewInternalError("read reserved field history sequence", err)
	}
	if len(reserved) != len(ordered) {
		return nil, historySerializationConflict("reserved field history sequence set is incomplete", nil)
	}
	return reserved, nil
}

// prelockHistorySequenceRows acquires the complete discovered sequence-row
// lock set in one deterministic multi-row statement without consuming a
// history sequence. It is only called by a root transaction retry owner on a
// fresh attempt.
func prelockHistorySequenceRows(ctx context.Context, tx interface{}, fieldIDs []string) error {
	exec, ok := tx.(Executor)
	if !ok || exec == nil {
		return apperrors.NewInternalError("history sequence prelock requires an active transaction executor", nil)
	}
	if _, rootDB := tx.(*sql.DB); rootDB {
		return apperrors.NewInternalError("history sequence prelock rejects the root database executor", nil)
	}

	ordered := ports.OrderedHistorySequenceFields(fieldIDs)
	if len(ordered) == 0 {
		return nil
	}
	valuePlaceholders := make([]string, len(ordered))
	args := make([]interface{}, len(ordered))
	for i, fieldID := range ordered {
		ports.MarkHistorySerializationAttempt(ctx)
		valuePlaceholders[i] = "(?, 0)"
		args[i] = fieldID
	}
	if _, err := exec.ExecContext(ctx, `
		INSERT INTO _System_FieldHistorySequence
			(FieldId, CurrentSequence)
		VALUES `+strings.Join(valuePlaceholders, ", ")+`
		ON DUPLICATE KEY UPDATE
			CurrentSequence = CurrentSequence`, args...); err != nil {
		if isOptimisticWriteConflict(err) {
			return historySerializationConflict("prelock field history sequence", err)
		}
		return apperrors.NewInternalError("prelock field history sequence", err)
	}
	for _, fieldID := range ordered {
		ports.MarkHistorySequenceLockAcquired(ctx, fieldID)
	}
	return nil
}

func historySerializationConflict(message string, cause error) error {
	if cause == nil {
		return fmt.Errorf("%w: %s", ports.ErrHistorySerializationConflict, message)
	}
	return fmt.Errorf("%w: %s: %v", ports.ErrHistorySerializationConflict, message, cause)
}