package persistence

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

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

// PermissionAuthoringConvergenceRepository owns durable post-commit recovery
// dispositions independently from command-result retention. Enqueue still
// joins the aggregate transaction and proves its immutable source command,
// but delivery can outlive expiry and recycling of that command's result.
type PermissionAuthoringConvergenceRepository struct {
	db        *sql.DB
	wake      func()
	ownership *RuntimeWorkOwnershipRepository
}

var permissionAuthoringConvergenceRequiredColumns = map[string]string{
	"SourceCommandRowId":       "varchar",
	"CommandId":                "varchar",
	"CommandDigest":            "char",
	"CanonicalPermissionSetId": "varchar",
	"AuthoringRevision":        "decimal(20,0)",
	"State":                    "varchar",
	"AttemptCount":             "bigint",
	"AvailableAt":              "datetime",
	"LeaseOwner":               "varchar",
	"LeaseExpiresAt":           "datetime",
	"LastError":                "mediumtext",
	"CreatedDate":              "datetime",
}

// VerifyPermissionAuthoringRuntimeSchema proves both durable halves needed
// before the API can accept an aggregate command. Worker-family ownership is a
// separate startup fence and must also be verified by process composition.
func VerifyPermissionAuthoringRuntimeSchema(ctx context.Context, db *sql.DB) error {
	if err := VerifyPermissionAuthoringCommandLifecycleSchema(ctx, db); err != nil {
		return err
	}
	const table = "_System_PermissionAuthoringConvergence"
	if err := VerifyRequiredTableColumns(ctx, db, table, permissionAuthoringConvergenceRequiredColumns); err != nil {
		return fmt.Errorf("permission authoring convergence: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		table,
		"ux_permission_authoring_convergence_source",
		true,
		[]string{"SourceCommandRowId"},
	); err != nil {
		return fmt.Errorf("permission authoring convergence: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		table,
		"idx_permission_authoring_convergence_delivery",
		false,
		[]string{"State", "AvailableAt", "Id"},
	); err != nil {
		return fmt.Errorf("permission authoring convergence: %w", err)
	}
	return nil
}

var _ ports.PermissionAuthoringConvergence = (*PermissionAuthoringConvergenceRepository)(nil)
var _ ports.PermissionAuthoringConvergenceStore = (*PermissionAuthoringConvergenceRepository)(nil)

type PermissionAuthoringConvergenceState struct {
	PendingCount int64
	OldestAge    time.Duration
}

func NewPermissionAuthoringConvergenceRepository(db *sql.DB, wake func()) (*PermissionAuthoringConvergenceRepository, error) {
	if db == nil {
		return nil, errors.New("permission authoring convergence database is required")
	}
	return &PermissionAuthoringConvergenceRepository{db: db, wake: wake}, nil
}

// NewFencedPermissionAuthoringConvergenceRepository is the independently
// deployed recovery-side owner. Every claim locks the active work-family fence
// in the same pessimistic transaction that takes row leases, so ownership loss
// cannot race one last unowned delivery batch.
func NewFencedPermissionAuthoringConvergenceRepository(
	db *sql.DB,
	expected RuntimeWorkOwnership,
) (*PermissionAuthoringConvergenceRepository, error) {
	if expected.WorkFamily != RuntimePermissionAuthoringCommandLifecycleWorkFamily {
		return nil, fmt.Errorf("permission authoring convergence: wrong work family %q", expected.WorkFamily)
	}
	ownership, err := NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	return &PermissionAuthoringConvergenceRepository{db: db, ownership: ownership}, nil
}

func (r *PermissionAuthoringConvergenceRepository) EnqueuePermissionAuthoringConvergence(
	ctx context.Context,
	event ports.PermissionAuthoringConvergenceEvent,
) error {
	if r == nil || r.db == nil {
		return errors.New("permission authoring convergence repository is unavailable")
	}
	if strings.TrimSpace(event.CommandRowID) == "" || strings.TrimSpace(event.CommandID) == "" ||
		!permissionAuthoringRequestDigest.MatchString(event.CommandDigest) ||
		strings.TrimSpace(event.CanonicalPermissionSet) == "" || event.AuthoringRevision == 0 {
		return errors.New("permission authoring convergence event is incomplete")
	}
	exec, err := (&PermissionAuthoringCommandRepository{db: r.db}).transactionExecutor(ctx)
	if err != nil {
		return err
	}
	result, err := exec.ExecContext(ctx, `INSERT INTO _System_PermissionAuthoringConvergence
(Id, SourceCommandRowId, CommandId, CommandDigest, CanonicalPermissionSetId,
 AuthoringRevision, State, AttemptCount, AvailableAt)
SELECT Id, Id, CommandId, RequestDigest, CanonicalPermissionSetId,
       ?, 'PENDING', 0, UTC_TIMESTAMP(6)
FROM _System_PermissionAuthoringCommand
WHERE Id = ? AND CommandId = ? AND RequestDigest = ?
  AND CanonicalPermissionSetId = ? AND State = 'ACCEPTED'`,
		event.AuthoringRevision, event.CommandRowID, event.CommandID,
		event.CommandDigest, event.CanonicalPermissionSet)
	if err != nil {
		return fmt.Errorf("enqueue permission authoring convergence: %w", err)
	}
	return requireOneAffected(result, "enqueue permission authoring convergence")
}

func (r *PermissionAuthoringConvergenceRepository) WakePermissionAuthoringConvergence() {
	if r != nil && r.wake != nil {
		r.wake()
	}
}

// ObservePermissionAuthoringConvergence reports the complete durable backlog,
// including leased and delayed-retry rows. CreatedDate preserves the true age
// of repeatedly retried work instead of making every failed attempt look new.
func (r *PermissionAuthoringConvergenceRepository) ObservePermissionAuthoringConvergence(
	ctx context.Context,
) (PermissionAuthoringConvergenceState, error) {
	if r == nil || r.db == nil {
		return PermissionAuthoringConvergenceState{}, errors.New("permission authoring convergence repository is unavailable")
	}
	var state PermissionAuthoringConvergenceState
	var oldestMicros int64
	if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*),
       COALESCE(GREATEST(TIMESTAMPDIFF(MICROSECOND, MIN(CreatedDate), UTC_TIMESTAMP(6)), 0), 0)
FROM _System_PermissionAuthoringConvergence
WHERE State = 'PENDING'`).Scan(&state.PendingCount, &oldestMicros); err != nil {
		return PermissionAuthoringConvergenceState{}, fmt.Errorf("observe permission authoring convergence: %w", err)
	}
	if state.PendingCount < 0 || oldestMicros < 0 {
		return PermissionAuthoringConvergenceState{}, errors.New("observe permission authoring convergence returned invalid state")
	}
	state.OldestAge = time.Duration(oldestMicros) * time.Microsecond
	return state, nil
}

func (r *PermissionAuthoringConvergenceRepository) ClaimPermissionAuthoringConvergence(
	ctx context.Context,
	leaseOwner string,
	leaseDuration time.Duration,
	limit int,
) ([]ports.PermissionAuthoringConvergenceWork, error) {
	leaseOwner = strings.TrimSpace(leaseOwner)
	if r == nil || r.db == nil || leaseOwner == "" || len(leaseOwner) > 255 ||
		leaseDuration <= 0 || leaseDuration.Microseconds() <= 0 || limit <= 0 || limit > 1000 {
		return nil, errors.New("permission authoring convergence claim is invalid")
	}
	if r.ownership == nil {
		return r.claimPermissionAuthoringConvergence(ctx, r.db, leaseOwner, leaseDuration, limit)
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return nil, fmt.Errorf("claim permission authoring convergence: begin: %w", err)
	}
	defer func() { _ = cleanup() }()
	if err := r.ownership.LockActive(ctx, tx); err != nil {
		_ = tx.Rollback()
		return nil, err
	}
	work, err := r.claimPermissionAuthoringConvergence(ctx, tx, leaseOwner, leaseDuration, limit)
	if err != nil {
		_ = tx.Rollback()
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("claim permission authoring convergence: commit outcome unknown: %w", err)
	}
	return work, nil
}

type permissionAuthoringConvergenceExecutor interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}

func (r *PermissionAuthoringConvergenceRepository) claimPermissionAuthoringConvergence(
	ctx context.Context,
	executor permissionAuthoringConvergenceExecutor,
	leaseOwner string,
	leaseDuration time.Duration,
	limit int,
) ([]ports.PermissionAuthoringConvergenceWork, error) {
	result, err := executor.ExecContext(ctx, `UPDATE _System_PermissionAuthoringConvergence
SET LeaseOwner = ?,
    LeaseExpiresAt = TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)),
    AttemptCount = AttemptCount + 1,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE State = 'PENDING'
  AND AvailableAt <= UTC_TIMESTAMP(6)
  AND (LeaseExpiresAt IS NULL OR LeaseExpiresAt <= UTC_TIMESTAMP(6))
ORDER BY AvailableAt, Id
LIMIT ?`, leaseOwner, leaseDuration.Microseconds(), limit)
	if err != nil {
		return nil, fmt.Errorf("claim permission authoring convergence: %w", err)
	}
	if _, err := result.RowsAffected(); err != nil {
		return nil, fmt.Errorf("claim permission authoring convergence rows affected: %w", err)
	}
	rows, err := executor.QueryContext(ctx, `SELECT SourceCommandRowId, CommandId, CommandDigest,
       CanonicalPermissionSetId, AuthoringRevision, AttemptCount
FROM _System_PermissionAuthoringConvergence
WHERE State = 'PENDING'
  AND LeaseOwner = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)
ORDER BY AvailableAt, Id`, leaseOwner)
	if err != nil {
		return nil, fmt.Errorf("load claimed permission authoring convergence: %w", err)
	}
	defer rows.Close()
	var work []ports.PermissionAuthoringConvergenceWork
	for rows.Next() {
		var item ports.PermissionAuthoringConvergenceWork
		if err := rows.Scan(
			&item.Event.CommandRowID,
			&item.Event.CommandID,
			&item.Event.CommandDigest,
			&item.Event.CanonicalPermissionSet,
			&item.Event.AuthoringRevision,
			&item.AttemptCount,
		); err != nil {
			return nil, fmt.Errorf("scan claimed permission authoring convergence: %w", err)
		}
		work = append(work, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate claimed permission authoring convergence: %w", err)
	}
	return work, nil
}

func (r *PermissionAuthoringConvergenceRepository) AcknowledgePermissionAuthoringConvergence(
	ctx context.Context,
	commandRowID string,
	leaseOwner string,
) error {
	if r == nil || r.db == nil || strings.TrimSpace(commandRowID) == "" || strings.TrimSpace(leaseOwner) == "" {
		return errors.New("permission authoring convergence acknowledgement is invalid")
	}
	return r.withPermissionAuthoringConvergenceOwnership(
		ctx,
		"acknowledge permission authoring convergence",
		func(executor permissionAuthoringConvergenceExecutor) error {
			result, err := executor.ExecContext(ctx, `DELETE FROM _System_PermissionAuthoringConvergence
WHERE SourceCommandRowId = ? AND State = 'PENDING'
  AND LeaseOwner = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`, commandRowID, leaseOwner)
			if err != nil {
				return fmt.Errorf("acknowledge permission authoring convergence: %w", err)
			}
			return requireOneAffected(result, "acknowledge permission authoring convergence")
		},
	)
}

func (r *PermissionAuthoringConvergenceRepository) RetryPermissionAuthoringConvergence(
	ctx context.Context,
	commandRowID string,
	leaseOwner string,
	lastError string,
	delay time.Duration,
) error {
	lastError = strings.TrimSpace(lastError)
	if r == nil || r.db == nil || strings.TrimSpace(commandRowID) == "" ||
		strings.TrimSpace(leaseOwner) == "" || lastError == "" || delay < 0 || delay.Microseconds() < 0 {
		return errors.New("permission authoring convergence retry is invalid")
	}
	if len(lastError) > 4096 {
		lastError = lastError[:4096]
	}
	return r.withPermissionAuthoringConvergenceOwnership(
		ctx,
		"retry permission authoring convergence",
		func(executor permissionAuthoringConvergenceExecutor) error {
			result, err := executor.ExecContext(ctx, `UPDATE _System_PermissionAuthoringConvergence
SET AvailableAt = TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)),
    LeaseOwner = NULL, LeaseExpiresAt = NULL,
    LastError = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE SourceCommandRowId = ? AND State = 'PENDING'
  AND LeaseOwner = ? AND LeaseExpiresAt > UTC_TIMESTAMP(6)`, delay.Microseconds(), lastError, commandRowID, leaseOwner)
			if err != nil {
				return fmt.Errorf("retry permission authoring convergence: %w", err)
			}
			return requireOneAffected(result, "retry permission authoring convergence")
		},
	)
}

func (r *PermissionAuthoringConvergenceRepository) withPermissionAuthoringConvergenceOwnership(
	ctx context.Context,
	operation string,
	mutate func(permissionAuthoringConvergenceExecutor) error,
) error {
	if r.ownership == nil {
		return mutate(r.db)
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return fmt.Errorf("%s: begin: %w", operation, err)
	}
	defer func() { _ = cleanup() }()
	if err := r.ownership.LockActive(ctx, tx); err != nil {
		_ = tx.Rollback()
		return err
	}
	if err := mutate(tx); err != nil {
		_ = tx.Rollback()
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("%s: commit outcome unknown: %w", operation, err)
	}
	return nil
}