package persistence

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

const RuntimePermissionAuthoringCommandLifecycleWorkFamily = "runtime.permission-authoring-command-lifecycle.v1"

var permissionAuthoringCommandLifecycleRequiredColumns = map[string]string{
	"State": "varchar", "CanonicalPermissionSetId": "varchar",
	"AdmissionExpiresAt": "datetime", "OutcomeExpiresAt": "datetime",
	"OutcomeRetentionMicros": "bigint", "SemanticMappingRevision": "bigint",
}

// PermissionAuthoringCommandLifecycleStore is the fenced lifecycle owner for
// the Permissions command journal. A nonterminal row proves that no business
// mutation committed: mutation and COMPLETED terminalization share one
// transaction. Expired nonterminal commands can therefore be abandoned by
// deletion after their row lock is acquired; completed commands are retired
// only after their replay deadline.
type PermissionAuthoringCommandLifecycleStore struct {
	db        *sql.DB
	ownership *RuntimeWorkOwnershipRepository
}

func NewPermissionAuthoringCommandLifecycleStore(
	db *sql.DB,
	expected RuntimeWorkOwnership,
) (*PermissionAuthoringCommandLifecycleStore, error) {
	ownership, err := NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	if expected.WorkFamily != RuntimePermissionAuthoringCommandLifecycleWorkFamily {
		return nil, fmt.Errorf("permission authoring command lifecycle: wrong work family %q", expected.WorkFamily)
	}
	return &PermissionAuthoringCommandLifecycleStore{db: db, ownership: ownership}, nil
}

func (s *PermissionAuthoringCommandLifecycleStore) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("permission authoring command lifecycle: store is unavailable")
	}
	if err := s.ownership.CheckReady(ctx); err != nil {
		return err
	}
	return VerifyPermissionAuthoringCommandLifecycleSchema(ctx, s.db)
}

func VerifyPermissionAuthoringCommandLifecycleSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return errors.New("permission authoring command lifecycle: context and database are required")
	}
	const table = "_System_PermissionAuthoringCommand"
	if err := VerifyRequiredTableColumns(ctx, db, table, permissionAuthoringCommandLifecycleRequiredColumns); err != nil {
		return fmt.Errorf("permission authoring command lifecycle: %w", err)
	}
	if err := VerifyExactTableIndex(ctx, db, table, "idx_permission_authoring_command_admission_expiry", false,
		[]string{"State", "AdmissionExpiresAt", "Id"}); err != nil {
		return fmt.Errorf("permission authoring command lifecycle: %w", err)
	}
	if err := VerifyExactTableIndex(ctx, db, table, "idx_permission_authoring_command_outcome_expiry", false,
		[]string{"State", "OutcomeExpiresAt", "Id"}); err != nil {
		return fmt.Errorf("permission authoring command lifecycle: %w", err)
	}
	return nil
}

// VerifyPermissionAuthoringSemanticCutover rejects startup while a live
// nonterminal command was accepted by another semantic planner. Deployment
// must first stop old admission and drain those commands; terminal rows remain
// replayable and new-revision rows remain recoverable across later restarts.
func VerifyPermissionAuthoringSemanticCutover(ctx context.Context, db *sql.DB, semanticMappingRevision uint32) error {
	if ctx == nil || db == nil || semanticMappingRevision == 0 {
		return errors.New("permission authoring semantic cutover: context, database, and revision are required")
	}
	var incompatibleActive bool
	if err := db.QueryRowContext(ctx, `SELECT EXISTS(
	SELECT 1 FROM _System_PermissionAuthoringCommand
	WHERE State = 'ACCEPTED' AND AdmissionExpiresAt > UTC_TIMESTAMP(6)
	  AND COALESCE(SemanticMappingRevision, 0) <> ?
)`, semanticMappingRevision).Scan(&incompatibleActive); err != nil {
		return fmt.Errorf("permission authoring semantic cutover: inspect active commands: %w", err)
	}
	if incompatibleActive {
		return errors.New("permission authoring semantic cutover: incompatible active commands must be drained before startup")
	}
	return nil
}

type PermissionAuthoringCommandLifecycleState struct {
	ExpiredNonterminalCount int64
	ExpiredTerminalCount    int64
	OldestAge               time.Duration
}

func (state PermissionAuthoringCommandLifecycleState) ExpiredCount() int64 {
	return state.ExpiredNonterminalCount + state.ExpiredTerminalCount
}

func (s *PermissionAuthoringCommandLifecycleStore) HasExpired(ctx context.Context) (bool, error) {
	if s == nil || s.db == nil {
		return false, errors.New("permission authoring command lifecycle: store is unavailable")
	}
	var exists bool
	if err := s.db.QueryRowContext(ctx, `SELECT EXISTS(
	SELECT 1 FROM _System_PermissionAuthoringCommand
	WHERE (State = ? AND AdmissionExpiresAt <= UTC_TIMESTAMP(6))
	   OR (State = ? AND OutcomeExpiresAt <= UTC_TIMESTAMP(6))
	LIMIT 1)`, PermissionAuthoringCommandAccepted,
		PermissionAuthoringCommandCompleted).Scan(&exists); err != nil {
		return false, fmt.Errorf("permission authoring command lifecycle: hint: %w", err)
	}
	return exists, nil
}

func (s *PermissionAuthoringCommandLifecycleStore) Observe(ctx context.Context) (PermissionAuthoringCommandLifecycleState, error) {
	if s == nil || s.db == nil {
		return PermissionAuthoringCommandLifecycleState{}, errors.New("permission authoring command lifecycle: store is unavailable")
	}
	var nonterminal, terminal, oldestMicros int64
	if err := s.db.QueryRowContext(ctx, `SELECT
	COALESCE(SUM(CASE WHEN State = ? AND AdmissionExpiresAt <= UTC_TIMESTAMP(6) THEN 1 ELSE 0 END), 0),
	COALESCE(SUM(CASE WHEN State = ? AND OutcomeExpiresAt <= UTC_TIMESTAMP(6) THEN 1 ELSE 0 END), 0),
	COALESCE(TIMESTAMPDIFF(MICROSECOND,
		MIN(CASE
			WHEN State = ? AND AdmissionExpiresAt <= UTC_TIMESTAMP(6) THEN AdmissionExpiresAt
			WHEN State = ? AND OutcomeExpiresAt <= UTC_TIMESTAMP(6) THEN OutcomeExpiresAt
		END), UTC_TIMESTAMP(6)), 0)
FROM _System_PermissionAuthoringCommand`,
		PermissionAuthoringCommandAccepted, PermissionAuthoringCommandCompleted,
		PermissionAuthoringCommandAccepted, PermissionAuthoringCommandCompleted,
	).Scan(&nonterminal, &terminal, &oldestMicros); err != nil {
		return PermissionAuthoringCommandLifecycleState{}, fmt.Errorf("permission authoring command lifecycle: observe: %w", err)
	}
	if nonterminal < 0 || terminal < 0 || oldestMicros < 0 {
		return PermissionAuthoringCommandLifecycleState{}, errors.New("permission authoring command lifecycle: invalid aggregate")
	}
	return PermissionAuthoringCommandLifecycleState{
		ExpiredNonterminalCount: nonterminal,
		ExpiredTerminalCount:    terminal,
		OldestAge:               time.Duration(oldestMicros) * time.Microsecond,
	}, nil
}

type PermissionAuthoringCommandCleanupResult struct {
	Abandoned int64
	Retired   int64
}

func (result PermissionAuthoringCommandCleanupResult) Processed() int64 {
	return result.Abandoned + result.Retired
}

// CleanupExpired performs one bounded, fenced, pessimistic transaction. The
// row lock serializes cleanup with an in-flight Save. If that Save commits,
// the row is COMPLETED and its fresh outcome deadline fails the guarded delete;
// if it rolls back, the expired nonterminal row is safe to abandon.
func (s *PermissionAuthoringCommandLifecycleStore) CleanupExpired(
	ctx context.Context,
	batchSize int,
) (PermissionAuthoringCommandCleanupResult, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return PermissionAuthoringCommandCleanupResult{}, errors.New("permission authoring command lifecycle: store is unavailable")
	}
	if batchSize <= 0 || batchSize > 10_000 {
		return PermissionAuthoringCommandCleanupResult{}, errors.New("permission authoring command lifecycle: invalid cleanup bounds")
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return PermissionAuthoringCommandCleanupResult{}, fmt.Errorf("permission authoring command lifecycle: begin: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (PermissionAuthoringCommandCleanupResult, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return PermissionAuthoringCommandCleanupResult{}, errors.Join(cause,
				fmt.Errorf("permission authoring command lifecycle: rollback: %w", rollbackErr))
		}
		return PermissionAuthoringCommandCleanupResult{}, cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	rows, err := tx.QueryContext(ctx, `SELECT Id, State FROM _System_PermissionAuthoringCommand
WHERE (State = 'ACCEPTED' AND AdmissionExpiresAt <= UTC_TIMESTAMP(6))
   OR (State = 'COMPLETED' AND OutcomeExpiresAt <= UTC_TIMESTAMP(6))
ORDER BY CASE WHEN State = 'COMPLETED' THEN OutcomeExpiresAt ELSE AdmissionExpiresAt END, Id
LIMIT ? FOR UPDATE`, batchSize)
	if err != nil {
		return rollback(fmt.Errorf("permission authoring command lifecycle: select expired: %w", err))
	}
	ids := make([]string, 0, batchSize)
	result := PermissionAuthoringCommandCleanupResult{}
	for rows.Next() {
		var id, state string
		if err := rows.Scan(&id, &state); err != nil {
			_ = rows.Close()
			return rollback(fmt.Errorf("permission authoring command lifecycle: scan expired: %w", err))
		}
		ids = append(ids, id)
		switch state {
		case PermissionAuthoringCommandAccepted:
			result.Abandoned++
		case PermissionAuthoringCommandCompleted:
			result.Retired++
		default:
			_ = rows.Close()
			return rollback(fmt.Errorf("permission authoring command lifecycle: selected unknown state %q", state))
		}
	}
	if err := rows.Close(); err != nil {
		return rollback(fmt.Errorf("permission authoring command lifecycle: close expired rows: %w", err))
	}
	if err := rows.Err(); err != nil {
		return rollback(fmt.Errorf("permission authoring command lifecycle: iterate expired: %w", err))
	}
	if len(ids) > 0 {
		placeholders := strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",")
		args := make([]any, len(ids))
		for i := range ids {
			args[i] = ids[i]
		}
		deleteSQL := `DELETE FROM _System_PermissionAuthoringCommand WHERE Id IN (` + placeholders + `)
AND ((State = 'ACCEPTED' AND AdmissionExpiresAt <= UTC_TIMESTAMP(6))
  OR (State = 'COMPLETED' AND OutcomeExpiresAt <= UTC_TIMESTAMP(6)))`
		deleted, err := tx.ExecContext(ctx, deleteSQL, args...)
		if err != nil {
			return rollback(fmt.Errorf("permission authoring command lifecycle: delete expired: %w", err))
		}
		count, err := deleted.RowsAffected()
		if err != nil {
			return rollback(fmt.Errorf("permission authoring command lifecycle: deleted rows: %w", err))
		}
		if count != int64(len(ids)) {
			return rollback(fmt.Errorf("permission authoring command lifecycle: deleted %d of %d locked commands", count, len(ids)))
		}
	}
	if err := tx.Commit(); err != nil {
		return PermissionAuthoringCommandCleanupResult{}, fmt.Errorf("permission authoring command lifecycle: commit outcome unknown: %w", err)
	}
	return result, nil
}