package controlplane

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

	"github.com/nexuscrm/persistence"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

type applyAllSourceRevisionAdvanceExactIncarnationExec func(
	context.Context,
	string,
	...any,
) (sql.Result, error)

// applyAllSourceRevisionAdvanceExactIncarnationCommitError means every
// mutation callback completed but the client could not determine whether the
// transaction commit became durable. Callers must not apply inverse effects;
// the durable row is the authority for a later lock-held reread.
type applyAllSourceRevisionAdvanceExactIncarnationCommitError struct {
	err error
}

func (e *applyAllSourceRevisionAdvanceExactIncarnationCommitError) Error() string {
	if e == nil || e.err == nil {
		return "exact incarnation transaction commit outcome is unknown"
	}
	return "exact incarnation transaction commit outcome is unknown: " + e.err.Error()
}

func (e *applyAllSourceRevisionAdvanceExactIncarnationCommitError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.err
}

type applyAllSourceRevisionAdvanceExactIncarnationStore struct {
	tx                       *sql.Tx
	exec                     func(context.Context, string, ...any) (sql.Result, error)
	query                    func(context.Context, string, ...any) (*sql.Rows, error)
	queryRow                 func(context.Context, string, ...any) *sql.Row
	publish                  func(context.Context, string) ([]string, error)
	oldIncarnationQueueGuard string
}

func (s applyAllSourceRevisionAdvanceExactIncarnationStore) ExecContext(
	ctx context.Context,
	query string,
	args ...any,
) (sql.Result, error) {
	return s.exec(ctx, query, args...)
}

func (s applyAllSourceRevisionAdvanceExactIncarnationStore) QueryContext(
	ctx context.Context,
	query string,
	args ...any,
) (*sql.Rows, error) {
	return s.query(ctx, query, args...)
}

func (s applyAllSourceRevisionAdvanceExactIncarnationStore) QueryRowContext(
	ctx context.Context,
	query string,
	args ...any,
) *sql.Row {
	return s.queryRow(ctx, query, args...)
}

func (s applyAllSourceRevisionAdvanceExactIncarnationStore) publishSourceBackedConnectedAppsForPack(
	ctx context.Context,
	packName string,
) ([]string, error) {
	return s.publish(ctx, packName)
}

func (s applyAllSourceRevisionAdvanceExactIncarnationStore) queueGuard() string {
	return s.oldIncarnationQueueGuard
}

func withApplyAllSourceRevisionAdvanceExactIncarnation(
	ctx context.Context,
	db *sql.DB,
	row installedPackRow,
	category packs.Category,
	mutate func(applyAllSourceRevisionAdvanceExactIncarnationExec, string) error,
) error {
	if mutate == nil {
		return fmt.Errorf("exact incarnation mutation is required")
	}
	return withApplyAllSourceRevisionAdvanceExactIncarnationStore(
		ctx, db, row, category,
		func(store applyAllSourceRevisionAdvanceExactIncarnationStore) error {
			return mutate(store.ExecContext, store.queueGuard())
		},
	)
}

// withApplyAllSourceRevisionAdvanceExactIncarnationStore serializes one mutation of
// an admitted installed row against ordinary connector enqueue. The callback
// runs only after the exact old row is locked and its non-terminal queue is
// drained. Operator replay has a separate connector contract: a replay that
// linearizes after AppPack evolution retains its old companion snapshot and
// fails closed at drain rather than executing against the new incarnation.
func withApplyAllSourceRevisionAdvanceExactIncarnationStore(
	ctx context.Context,
	db *sql.DB,
	row installedPackRow,
	category packs.Category,
	mutate func(applyAllSourceRevisionAdvanceExactIncarnationStore) error,
) error {
	if db == nil {
		return fmt.Errorf("exact incarnation database is required")
	}
	if mutate == nil {
		return fmt.Errorf("exact incarnation mutation is required")
	}
	if category != packs.AppPack && category != packs.ConnectorPack {
		return fmt.Errorf("category %q has no exact queue-ownership rule", category)
	}
	if strings.TrimSpace(row.ID) == "" || strings.TrimSpace(row.PackInstallID) == "" {
		return fmt.Errorf("installed row and pack-install incarnation identity are required")
	}
	if row.Category != string(category) {
		return fmt.Errorf("installed row category %q does not match queue category %q", row.Category, category)
	}
	queueBlocker, err := resolveApplyAllSourceRevisionAdvanceOldIncarnationQueueBlocker(
		ctx, db, category,
	)
	if err != nil {
		return err
	}
	queueGuard := applyAllSourceRevisionAdvanceOldIncarnationQueueGuard(queueBlocker)
	if queueGuard == "" {
		return fmt.Errorf("category %q has no exact queue-ownership rule", category)
	}

	tx, cleanup, err := persistence.BeginPessimisticTx(
		persistence.WithReadCommittedIsolation(ctx), db,
	)
	if err != nil {
		return fmt.Errorf("begin exact incarnation transition: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	if err := lockApplyAllSourceRevisionAdvanceExactInstalledRow(ctx, tx, row); err != nil {
		return err
	}
	queued, err := hasApplyAllSourceRevisionAdvanceOldIncarnationQueue(
		ctx, tx, row, queueBlocker,
	)
	if err != nil {
		return err
	}
	if queued {
		return fmt.Errorf("prior pack-install incarnation queue is not drained")
	}
	store := applyAllSourceRevisionAdvanceExactIncarnationStore{
		tx:       tx,
		exec:     tx.ExecContext,
		query:    tx.QueryContext,
		queryRow: tx.QueryRowContext,
		publish: func(ctx context.Context, packName string) ([]string, error) {
			return packs.PublishSourceBackedConnectedAppsForPackInTx(ctx, tx, packName)
		},
		oldIncarnationQueueGuard: queueGuard,
	}
	if err := mutate(store); err != nil {
		return fmt.Errorf("mutate exact installed-row incarnation: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return &applyAllSourceRevisionAdvanceExactIncarnationCommitError{err: err}
	}
	committed = true
	return nil
}

func lockApplyAllSourceRevisionAdvanceExactInstalledRow(
	ctx context.Context,
	tx *sql.Tx,
	row installedPackRow,
) error {
	predicate, args := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
	var one int
	err := tx.QueryRowContext(ctx, `SELECT 1
  FROM _System_InstalledPack p
 WHERE `+predicate+`
 LIMIT 1
 FOR UPDATE`, args...).Scan(&one)
	if err == sql.ErrNoRows {
		return fmt.Errorf("frozen installed row changed before incarnation lock")
	}
	if err != nil {
		return fmt.Errorf("lock frozen installed row incarnation: %w", err)
	}
	return nil
}

func hasApplyAllSourceRevisionAdvanceOldIncarnationQueue(
	ctx context.Context,
	tx *sql.Tx,
	row installedPackRow,
	queueBlocker string,
) (bool, error) {
	if strings.TrimSpace(queueBlocker) == "" {
		return false, fmt.Errorf("exact queue blocker is required")
	}
	predicate, args := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
	var one int
	err := tx.QueryRowContext(ctx, `SELECT 1
  FROM _System_InstalledPack p
 WHERE `+predicate+`
   AND (`+queueBlocker+`)
 LIMIT 1`, args...).Scan(&one)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("read prior pack-install incarnation queue: %w", err)
	}
	return true, nil
}

func applyAllSourceRevisionAdvanceOldIncarnationQueueGuard(
	queueBlocker string,
) string {
	if strings.TrimSpace(queueBlocker) == "" {
		return ""
	}
	return `
   AND NOT (` + queueBlocker + `)`
}

func resolveApplyAllSourceRevisionAdvanceOldIncarnationQueueBlocker(
	ctx context.Context,
	db *sql.DB,
	category packs.Category,
) (string, error) {
	scope := connectorPackInboundReapplyScope
	switch category {
	case packs.ConnectorPack:
	case packs.AppPack:
		scope = companionAppPackInboundReapplyScope
	default:
		return "", fmt.Errorf("category %q has no exact queue-ownership rule", category)
	}
	workItemAuthority, err := connectorInboundWorkItemSchemaReady(ctx, db)
	if err != nil {
		return "", fmt.Errorf("resolve exact incarnation queue authority: %w", err)
	}
	return connectorInboundExactIncarnationBlockerPredicate(
		scope, workItemAuthority,
	), nil
}