package catalogtenant

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

	tenantauthorityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/tenantlayoutbundleauthorityv1"
)

// ObservedNamespaceReservation is one current tenant namespace reservation
// observed while the tenant authority fence is locked. It is only transient,
// stopped-state preflight inventory; it is not a namespace-owner claim, an
// issuer capability, or authorization to map a ProfileKey or RoleKey.
type ObservedNamespaceReservation struct {
	NamespaceID string
	State       tenantauthorityv1.NamespaceState
}

type namespaceReservationHistory struct {
	commandID tenantauthorityv1.CommandID
	sequence  uint64
	value     tenantauthorityv1.TenantNamespaceReservationV1
	digest    string
	prior     ValueReference
}

// InspectNamespaceReservationLineage opens a source-owned transaction, locks
// the tenant authority fence, and verifies the complete durable lifecycle of
// every current tenant namespace reservation: one ACTIVE reserve followed by
// an optional TOMBSTONE. It always rolls back before returning and cannot
// mutate or adopt namespace state. A later migration must independently
// re-inspect and revalidate its source facts in its own stopped-phase
// transaction before performing any mutation.
func (a *Authority) InspectNamespaceReservationLineage(
	ctx context.Context,
) (reservations []ObservedNamespaceReservation, returnErr error) {
	if a == nil || a.db == nil || a.identity == nil {
		return nil, ErrConflict
	}

	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() {
		rollbackErr := tx.Rollback()
		if rollbackErr != nil {
			reservations = nil
			returnErr = errors.Join(returnErr, rollbackErr)
		}
	}()
	sequence, err := lockAuthorityFence(ctx, tx)
	if err != nil {
		return nil, err
	}
	reservations, err = inspectNamespaceReservationLineage(ctx, tx, sequence)
	if err != nil {
		return nil, err
	}
	return reservations, nil
}

func inspectNamespaceReservationLineage(
	ctx context.Context,
	tx *sql.Tx,
	fenceSequence uint64,
) ([]ObservedNamespaceReservation, error) {
	currents, err := readNamespaceReservationCurrents(ctx, tx)
	if err != nil {
		return nil, err
	}
	histories, err := readNamespaceReservationHistories(ctx, tx)
	if err != nil {
		return nil, err
	}
	appliedCommands, err := readAppliedNamespaceCommands(ctx, tx)
	if err != nil {
		return nil, err
	}

	namespaceIDs := make([]string, 0, len(currents))
	for namespaceID := range currents {
		namespaceIDs = append(namespaceIDs, namespaceID)
	}
	sort.Strings(namespaceIDs)

	reservations := make([]ObservedNamespaceReservation, 0, len(currents))
	for _, namespaceID := range namespaceIDs {
		current := currents[namespaceID]
		history := histories[namespaceID]
		if err := validateNamespaceReservationLifecycle(
			ctx,
			tx,
			fenceSequence,
			current,
			history,
			appliedCommands,
		); err != nil {
			return nil, err
		}
		delete(histories, namespaceID)
		reservations = append(reservations, ObservedNamespaceReservation{
			NamespaceID: namespaceID,
			State:       current.Value.State,
		})
	}
	if len(histories) != 0 {
		return nil, fmt.Errorf("%w: namespace history has no current reservation", ErrCorrupt)
	}
	if len(appliedCommands) != 0 {
		return nil, fmt.Errorf("%w: applied namespace command has no history", ErrCorrupt)
	}
	return reservations, nil
}

func readNamespaceReservationCurrents(
	ctx context.Context,
	tx *sql.Tx,
) (map[string]namespaceCurrent, error) {
	rows, err := tx.QueryContext(ctx, `SELECT TenantNamespaceId, ValueVersion, State,
ValueDigest, LastCommandId
FROM _System_CatalogTenantNamespaceReservation
ORDER BY TenantNamespaceId FOR UPDATE`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	currents := make(map[string]namespaceCurrent)
	for rows.Next() {
		var current namespaceCurrent
		var storedNamespaceID []byte
		var rawState, storedCommandID string
		if err := rows.Scan(
			&storedNamespaceID,
			&current.Value.ValueVersion,
			&rawState,
			&current.Digest,
			&storedCommandID,
		); err != nil {
			return nil, err
		}
		current.Value.TenantNamespaceID = string(storedNamespaceID)
		current.Value.State = tenantauthorityv1.NamespaceState(rawState)
		current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
		if _, exists := currents[current.Value.TenantNamespaceID]; exists {
			return nil, fmt.Errorf("%w: duplicate namespace current", ErrCorrupt)
		}
		if err := validateNamespaceCurrent(current); err != nil {
			return nil, err
		}
		currents[current.Value.TenantNamespaceID] = current
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return currents, nil
}

func readNamespaceReservationHistories(
	ctx context.Context,
	tx *sql.Tx,
) (map[string][]namespaceReservationHistory, error) {
	rows, err := tx.QueryContext(ctx, `SELECT CommandId, CommandSequence,
TenantNamespaceId, ValueVersion, State, ValueDigest, PriorValueDigest
FROM _System_CatalogTenantNamespaceHistory
ORDER BY TenantNamespaceId, CommandSequence, CommandId FOR UPDATE`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	histories := make(map[string][]namespaceReservationHistory)
	commandIDs := make(map[tenantauthorityv1.CommandID]struct{})
	for rows.Next() {
		var history namespaceReservationHistory
		var storedCommandID, rawState string
		var rawSequence, storedNamespaceID []byte
		var prior sql.NullString
		if err := rows.Scan(
			&storedCommandID,
			&rawSequence,
			&storedNamespaceID,
			&history.value.ValueVersion,
			&rawState,
			&history.digest,
			&prior,
		); err != nil {
			return nil, err
		}
		history.commandID = tenantauthorityv1.CommandID(storedCommandID)
		if err := tenantauthorityv1.ValidateCommandID(history.commandID); err != nil {
			return nil, fmt.Errorf("%w: invalid namespace history command: %v", ErrCorrupt, err)
		}
		if _, exists := commandIDs[history.commandID]; exists {
			return nil, fmt.Errorf("%w: duplicate namespace history command", ErrCorrupt)
		}
		commandIDs[history.commandID] = struct{}{}
		history.sequence, err = parsePositiveDurableSequence(
			"namespace history CommandSequence",
			rawSequence,
		)
		if err != nil {
			return nil, err
		}
		history.value.TenantNamespaceID = string(storedNamespaceID)
		history.value.State = tenantauthorityv1.NamespaceState(rawState)
		history.prior, err = durableReference("namespace history PriorValueDigest", prior.Valid, prior)
		if err != nil {
			return nil, err
		}
		computed, err := tenantauthorityv1.NamespaceValueDigest(history.value)
		if err != nil {
			return nil, fmt.Errorf("%w: invalid namespace history value: %v", ErrCorrupt, err)
		}
		if computed != history.digest {
			return nil, fmt.Errorf("%w: namespace history digest differs", ErrCorrupt)
		}
		histories[history.value.TenantNamespaceID] = append(
			histories[history.value.TenantNamespaceID],
			history,
		)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return histories, nil
}

func readAppliedNamespaceCommands(
	ctx context.Context,
	tx *sql.Tx,
) (map[tenantauthorityv1.CommandID]uint64, error) {
	rows, err := tx.QueryContext(ctx, `SELECT CommandId, CommandSequence
FROM _System_CatalogTenantAuthorityCommand
WHERE CommandKind = ? AND Result = ?
ORDER BY CommandSequence, CommandId FOR UPDATE`, commandKindNamespace, ResultApplied)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	commands := make(map[tenantauthorityv1.CommandID]uint64)
	for rows.Next() {
		var storedCommandID string
		var rawSequence []byte
		if err := rows.Scan(&storedCommandID, &rawSequence); err != nil {
			return nil, err
		}
		commandID := tenantauthorityv1.CommandID(storedCommandID)
		if err := tenantauthorityv1.ValidateCommandID(commandID); err != nil {
			return nil, fmt.Errorf("%w: invalid applied namespace command: %v", ErrCorrupt, err)
		}
		sequence, err := parsePositiveDurableSequence(
			"applied namespace command CommandSequence",
			rawSequence,
		)
		if err != nil {
			return nil, err
		}
		if _, exists := commands[commandID]; exists {
			return nil, fmt.Errorf("%w: duplicate applied namespace command", ErrCorrupt)
		}
		commands[commandID] = sequence
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return commands, nil
}

func validateNamespaceReservationLifecycle(
	ctx context.Context,
	tx *sql.Tx,
	fenceSequence uint64,
	current namespaceCurrent,
	history []namespaceReservationHistory,
	appliedCommands map[tenantauthorityv1.CommandID]uint64,
) error {
	if len(history) == 0 || len(history) > 2 {
		return fmt.Errorf("%w: namespace reservation must have one reserve and optional tombstone", ErrCorrupt)
	}
	if history[0].value.State != tenantauthorityv1.NamespaceStateActive ||
		history[0].prior.Present {
		return fmt.Errorf("%w: namespace reservation must begin with active reserve", ErrCorrupt)
	}
	if len(history) == 2 {
		if history[1].value.State != tenantauthorityv1.NamespaceStateTombstone ||
			history[1].prior != (ValueReference{Present: true, Digest: history[0].digest}) ||
			history[0].sequence >= history[1].sequence {
			return fmt.Errorf("%w: namespace tombstone must follow its active reserve", ErrCorrupt)
		}
	}
	last := history[len(history)-1]
	if current.Value != last.value || current.Digest != last.digest ||
		current.LastCommandID != last.commandID {
		return fmt.Errorf("%w: namespace current differs from final history", ErrCorrupt)
	}

	for _, item := range history {
		if item.sequence > fenceSequence {
			return fmt.Errorf("%w: namespace history sequence exceeds authority fence", ErrCorrupt)
		}
		commandSequence, exists := appliedCommands[item.commandID]
		if !exists || commandSequence != item.sequence {
			return fmt.Errorf("%w: namespace history lacks its applied terminal command", ErrCorrupt)
		}
		command, err := readTerminalCommand(ctx, tx, item.commandID)
		if err != nil {
			return err
		}
		if command.Kind != commandKindNamespace || command.Result != ResultApplied ||
			command.Sequence != item.sequence || command.Sequence > fenceSequence {
			return fmt.Errorf("%w: namespace terminal differs from history", ErrCorrupt)
		}
		request, err := namespaceHistoryRequest(item)
		if err != nil {
			return err
		}
		if err := validateNamespaceOutcome(request, command); err != nil {
			return fmt.Errorf("%w: namespace terminal does not validate its history: %v", ErrCorrupt, err)
		}
		delete(appliedCommands, item.commandID)
	}
	return nil
}

func namespaceHistoryRequest(
	history namespaceReservationHistory,
) (tenantauthorityv1.TenantNamespaceCommandRequest, error) {
	request := tenantauthorityv1.TenantNamespaceCommandRequest{
		TargetTenantNamespaceID: history.value.TenantNamespaceID,
		DesiredNamespaceValue:   history.value,
	}
	switch history.value.State {
	case tenantauthorityv1.NamespaceStateActive:
		if history.prior.Present {
			return tenantauthorityv1.TenantNamespaceCommandRequest{}, fmt.Errorf(
				"%w: active reserve must not carry a prior value", ErrCorrupt,
			)
		}
		request.Operation = tenantauthorityv1.NamespaceOperationReserve
		request.ExpectedPriorNamespace = tenantauthorityv1.TenantNamespaceExpectedPrior{
			Variant: tenantauthorityv1.NamespaceExpectedPriorAbsent,
		}
	case tenantauthorityv1.NamespaceStateTombstone:
		if !history.prior.Present {
			return tenantauthorityv1.TenantNamespaceCommandRequest{}, fmt.Errorf(
				"%w: tombstone must carry its active prior value", ErrCorrupt,
			)
		}
		request.Operation = tenantauthorityv1.NamespaceOperationTombstone
		request.ExpectedPriorNamespace = tenantauthorityv1.TenantNamespaceExpectedPrior{
			Variant:              tenantauthorityv1.NamespaceExpectedPriorExact,
			NamespaceValueDigest: history.prior.Digest,
		}
	default:
		return tenantauthorityv1.TenantNamespaceCommandRequest{}, fmt.Errorf(
			"%w: namespace history state is invalid", ErrCorrupt,
		)
	}
	return request, nil
}

func parsePositiveDurableSequence(label string, raw []byte) (uint64, error) {
	value, err := parseDurableUint64(label, raw)
	if err != nil {
		return 0, err
	}
	if value == 0 || string(raw) != uintText(value) {
		return 0, fmt.Errorf("%w: %s is not a positive canonical unsigned decimal", ErrCorrupt, label)
	}
	return value, nil
}