package admission

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

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

const (
	ModeActive      = "active"
	ModeQuiescing   = "quiescing"
	ModePaused      = "paused"
	ModeUnavailable = "unavailable"
)

// Capability describes whether the L0 admission authority can safely decide a
// namespace boundary. A missing control-plane table is unavailable, never an
// implicit allow.
type Capability string

const (
	CapabilityAvailable   Capability = "available"
	CapabilityUnavailable Capability = "unavailable"
)

type Decision struct {
	Allow             bool
	Mode              string
	RetryAfterSeconds int
	Reason            string
}

// Fence is the durable, namespace-wide ingress boundary. Its row is scoped by
// (Namespace, PackInstallId="", Channel="*"). FenceEpoch is an ABA-safe value:
// the fence writer increments it once for every active-to-quiescing transition
// and never reuses a prior value.
type Fence struct {
	Namespace         string
	Mode              string
	RetryAfterSeconds int
	Reason            string
	Epoch             int64
	Capability        Capability
}

// ErrIngressQuiesced is returned when the transaction-authoritative namespace
// fence rejects new Event and WorkItem materialization. Callers should retry
// rather than acknowledge the provider delivery.
var (
	ErrIngressQuiesced          = errors.New("connector admission: namespace ingress is quiesced")
	ErrIngressSchemaUnavailable = errors.New("connector admission: ingress-control schema is unavailable")
)

// SchemaUnavailableError is returned when the required control-plane table or
// FenceEpoch column is unavailable, or when the table cannot establish one
// exact canonical namespace fence. It is deliberately typed so transports map
// it to controlled 503 and callers never materialize an Event or WorkItem.
type SchemaUnavailableError struct {
	Namespace string
	Cause     error
}

func (e *SchemaUnavailableError) Error() string {
	if e == nil {
		return ErrIngressSchemaUnavailable.Error()
	}
	if e.Cause == nil {
		return fmt.Sprintf("%s: namespace=%q", ErrIngressSchemaUnavailable, e.Namespace)
	}
	return fmt.Sprintf("%s: namespace=%q: %v", ErrIngressSchemaUnavailable, e.Namespace, e.Cause)
}

func (e *SchemaUnavailableError) Unwrap() error { return e.Cause }

func (e *SchemaUnavailableError) Is(target error) bool { return target == ErrIngressSchemaUnavailable }

// QuiescedError carries the committed fence state that rejected a materialize
// transaction. FenceEpoch lets release receipts distinguish pause-resume-pause
// ABA cycles even when the later mode string is identical to an earlier one.
type QuiescedError struct {
	Fence Fence
}

func (e *QuiescedError) Error() string {
	if e == nil {
		return ErrIngressQuiesced.Error()
	}
	return fmt.Sprintf("%s: namespace=%q mode=%q epoch=%d", ErrIngressQuiesced, e.Fence.Namespace, e.Fence.Mode, e.Fence.Epoch)
}

func (e *QuiescedError) Is(target error) bool {
	return target == ErrIngressQuiesced
}

type Gate interface {
	Decide(ctx context.Context, namespace, packInstallID, channel string) (Decision, error)
}

// NoopGate and NewNoopGate remain part of the published module surface for
// pinned consumers that deliberately opt out of database-backed preflight.
// Durable ingress writers use TransactionGate instead; retaining this type is
// not permission to bypass RequireActiveTx on the production write path.
type NoopGate struct{}

func NewNoopGate() Gate {
	return NoopGate{}
}

func (NoopGate) Decide(context.Context, string, string, string) (Decision, error) {
	return Allow(), nil
}

// TransactionGate is the authority for durable inbound materialization. The
// caller must invoke it after opening, and before writing anything in, the same
// pessimistic transaction that creates _System_ConnectorEvent and its WorkItem.
// Decide remains a best-effort preflight only; it cannot decide admission.
type TransactionGate interface {
	RequireActiveTx(ctx context.Context, tx *sql.Tx, namespace string) (Fence, error)
}

// NamespaceFenceTransition is the only L0 write request for the stable
// namespace fence. Callers cannot choose an epoch; Arm advances it exactly once
// on active -> quiescing and Resume never resets it.
type NamespaceFenceTransition struct {
	Namespace         string
	TargetMode        string
	RetryAfterSeconds int
	Reason            string
}

// FenceReceipt is the receipt-safe identity of a committed quiesce. Epoch zero
// is deliberately invalid because it predates the receipt-safe transition
// contract.
type FenceReceipt struct {
	Namespace string
	Epoch     int64
}

// CommittedReceipt accepts only a committed, quiescing fence with an epoch that
// cannot be confused with the pre-epoch baseline. TransitionNamespaceFence
// invokes it only after its transaction commits.
func (f Fence) CommittedReceipt() (FenceReceipt, error) {
	if f.Capability != CapabilityAvailable || canonicalNamespace(f.Namespace) == "" || f.Mode != ModeQuiescing || f.Epoch <= 0 {
		return FenceReceipt{}, fmt.Errorf("connector admission: fence is not receipt-safe: namespace=%q mode=%q epoch=%d capability=%q", f.Namespace, f.Mode, f.Epoch, f.Capability)
	}
	return FenceReceipt{Namespace: canonicalNamespace(f.Namespace), Epoch: f.Epoch}, nil
}

// TransitionResult is returned only after the namespace-fence transaction has
// committed. Receipt is populated for quiescing transitions and rejects epoch 0.
type TransitionResult struct {
	Fence   Fence
	Receipt *FenceReceipt
}

type SQLGate struct {
	db  *sql.DB
	now func() time.Time
}

func NewSQLGate(db *sql.DB) *SQLGate {
	return &SQLGate{db: db, now: time.Now}
}

func (g *SQLGate) Decide(ctx context.Context, namespace, packInstallID, channel string) (Decision, error) {
	if g == nil || g.db == nil {
		return unavailableDecision(), schemaUnavailable(namespace, nil)
	}
	namespace = strings.ToLower(strings.TrimSpace(namespace))
	packInstallID = strings.TrimSpace(packInstallID)
	channel = strings.ToLower(strings.TrimSpace(channel))
	if namespace == "" || packInstallID == "" {
		return Allow(), nil
	}

	channels := channelCandidates(channel)
	args := []interface{}{namespace, packInstallID}
	query := `
SELECT Mode, COALESCE(RetryAfterSeconds, 0), COALESCE(Reason, ''), COALESCE(FenceEpoch, 0)
  FROM _System_ConnectorIngressControl
 WHERE Namespace = ?
   AND (PackInstallId = ? OR COALESCE(PackInstallId, '') = '')
   AND Channel IN (` + placeholders(len(channels)) + `)
   AND COALESCE(IsDeleted, 0) = 0
   AND (ExpiresAt IS NULL OR ExpiresAt > ?)
 ORDER BY
   CASE WHEN PackInstallId = ? THEN 0 ELSE 1 END,
   CASE Channel
`
	args = append(args, stringArgs(channels)...)
	args = append(args, g.now().UTC(), packInstallID)
	for i, candidate := range channels {
		query += fmt.Sprintf("     WHEN ? THEN %d\n", i)
		args = append(args, candidate)
	}
	query += `     ELSE 99
   END,
   LastModifiedDate DESC
 LIMIT 1`

	var (
		mode       string
		retryAfter int
		reason     string
		epoch      int64
	)
	// FenceEpoch is selected even though Decide does not use its value to decide
	// a mode. This fast read must prove the receipt-capable schema exists before
	// it can advertise CapabilityAvailable; a pre-epoch table fails closed.
	err := g.db.QueryRowContext(ctx, query, args...).Scan(&mode, &retryAfter, &reason, &epoch)
	if err == sql.ErrNoRows {
		return Allow(), nil
	}
	if err != nil {
		if isMissingIngressControlTable(err) || isMissingFenceEpoch(err) {
			return unavailableDecision(), schemaUnavailable(namespace, err)
		}
		return Decision{}, fmt.Errorf("connector admission: query ingress control %s/%s: %w", namespace, channel, err)
	}

	mode = strings.ToLower(strings.TrimSpace(mode))
	if mode == "" || mode == ModeActive {
		return Allow(), nil
	}
	if retryAfter <= 0 {
		retryAfter = 120
	}
	return Decision{
		Allow:             false,
		Mode:              mode,
		RetryAfterSeconds: retryAfter,
		Reason:            strings.TrimSpace(reason),
	}, nil
}

// RequireActiveTx locks the stable namespace fence in the caller's pessimistic
// transaction. It deliberately ignores pack/channel-specific fast-gate rows:
// only the global namespace row can serialize a release quiesce against every
// Event + WorkItem materialization path.
//
// The INSERT ... ON DUPLICATE KEY UPDATE creates an active baseline exactly
// once and takes the unique (Namespace, PackInstallId, Channel) row lock even
// when another writer created the row. A release fence update therefore either
// commits before this method reads it (and rejects this transaction), or waits
// for this admitted transaction to commit before it can arm. There is no
// check-then-insert window.
func (g *SQLGate) RequireActiveTx(ctx context.Context, tx *sql.Tx, namespace string) (Fence, error) {
	fence, err := g.lockNamespaceFence(ctx, tx, namespace)
	if err != nil {
		return Fence{}, err
	}
	if fence.Mode == ModeActive {
		return fence, nil
	}
	if fence.RetryAfterSeconds <= 0 {
		fence.RetryAfterSeconds = 120
	}
	return Fence{}, &QuiescedError{Fence: fence}
}

// TransitionNamespaceFence is the single L0 arm/resume write path. It locks
// the same global row as RequireActiveTx in a pessimistic transaction, so an
// arm cannot pass a previously admitted Event/WorkItem or Replay materializer.
func (g *SQLGate) TransitionNamespaceFence(ctx context.Context, request NamespaceFenceTransition) (TransitionResult, error) {
	namespace := canonicalNamespace(request.Namespace)
	if g == nil || g.db == nil || namespace == "" {
		return TransitionResult{}, schemaUnavailable(namespace, nil)
	}
	target := strings.ToLower(strings.TrimSpace(request.TargetMode))
	if target != ModeActive && target != ModeQuiescing {
		return TransitionResult{}, fmt.Errorf("connector admission: unsupported namespace fence target mode %q", request.TargetMode)
	}

	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, g.db)
	if err != nil {
		return TransitionResult{}, err
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	fence, err := g.lockNamespaceFence(ctx, tx, namespace)
	if err != nil {
		return TransitionResult{}, err
	}

	switch target {
	case ModeQuiescing:
		if fence.Mode == ModeActive {
			retryAfter := request.RetryAfterSeconds
			if retryAfter <= 0 {
				retryAfter = 120
			}
			result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorIngressControl
   SET Mode = ?, FenceEpoch = FenceEpoch + 1, RetryAfterSeconds = ?, Reason = ?, ExpiresAt = NULL, LastModifiedDate = CURRENT_TIMESTAMP
	 WHERE Namespace = ? AND PackInstallId = '' AND Channel = '*' AND IsDeleted = 0 AND FenceEpoch IS NOT NULL`,
				ModeQuiescing, retryAfter, strings.TrimSpace(request.Reason), namespace)
			if err != nil {
				return TransitionResult{}, fmt.Errorf("connector admission: arm namespace fence %q: %w", namespace, err)
			}
			if err := requireOneCanonicalFenceUpdate(namespace, result); err != nil {
				return TransitionResult{}, err
			}
			fence.Mode = ModeQuiescing
			fence.RetryAfterSeconds = retryAfter
			fence.Reason = strings.TrimSpace(request.Reason)
			fence.Epoch++
		} else if fence.Mode != ModeQuiescing {
			return TransitionResult{}, fmt.Errorf("connector admission: cannot arm namespace fence %q from mode %q", namespace, fence.Mode)
		}
	case ModeActive:
		if fence.Mode != ModeActive {
			result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorIngressControl
   SET Mode = ?, Reason = '', ExpiresAt = NULL, LastModifiedDate = CURRENT_TIMESTAMP
	 WHERE Namespace = ? AND PackInstallId = '' AND Channel = '*' AND IsDeleted = 0 AND FenceEpoch IS NOT NULL`,
				ModeActive, namespace)
			if err != nil {
				return TransitionResult{}, fmt.Errorf("connector admission: resume namespace fence %q: %w", namespace, err)
			}
			if err := requireOneCanonicalFenceUpdate(namespace, result); err != nil {
				return TransitionResult{}, err
			}
			fence.Mode = ModeActive
			fence.Reason = ""
		}
	}

	if err := tx.Commit(); err != nil {
		return TransitionResult{}, err
	}
	committed = true
	result := TransitionResult{Fence: fence}
	if target == ModeQuiescing {
		receipt, err := fence.CommittedReceipt()
		if err != nil {
			return TransitionResult{}, err
		}
		result.Receipt = &receipt
	}
	return result, nil
}

func (g *SQLGate) lockNamespaceFence(ctx context.Context, tx *sql.Tx, namespace string) (Fence, error) {
	namespace = canonicalNamespace(namespace)
	if g == nil || g.db == nil || namespace == "" {
		return Fence{}, schemaUnavailable(namespace, nil)
	}
	if tx == nil {
		return Fence{}, errors.New("connector admission: transaction fence requires a transaction")
	}

	_, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorIngressControl
  (Id, Namespace, PackInstallId, Channel, Mode, FenceEpoch, RetryAfterSeconds, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, '', '*', ?, 0, 120, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON DUPLICATE KEY UPDATE
  Mode = IF(IsDeleted <> 0, 'active', Mode),
  RetryAfterSeconds = IF(IsDeleted <> 0, 120, RetryAfterSeconds),
  Reason = IF(IsDeleted <> 0, NULL, Reason),
  ExpiresAt = IF(IsDeleted <> 0, NULL, ExpiresAt),
  LastModifiedDate = IF(IsDeleted <> 0, CURRENT_TIMESTAMP, LastModifiedDate),
  IsDeleted = 0`, NamespaceFenceID(namespace), namespace, ModeActive)
	if err != nil {
		if isMissingIngressControlTable(err) {
			return Fence{}, schemaUnavailable(namespace, err)
		}
		if isMissingFenceEpoch(err) {
			return Fence{}, schemaUnavailable(namespace, err)
		}
		return Fence{}, fmt.Errorf("connector admission: establish namespace fence %q: %w", namespace, err)
	}

	// The upsert above serializes creation on the exact canonical key. We then
	// inspect every active logical-global row, without COALESCE or LIMIT: a
	// legacy NULL scope, NULL epoch, or duplicate row is not a state the
	// runtime may normalize or select arbitrarily. Controlplane owns repair;
	// Connector admission fails closed until that repair has converged it.
	rows, err := tx.QueryContext(ctx, `
SELECT PackInstallId, Mode, COALESCE(RetryAfterSeconds, 0), COALESCE(Reason, ''), FenceEpoch, ExpiresAt
  FROM _System_ConnectorIngressControl
 WHERE Namespace = ?
   AND (PackInstallId IS NULL OR PackInstallId = '')
   AND Channel = '*'
   AND COALESCE(IsDeleted, 0) = 0
 FOR UPDATE`, namespace)
	if err != nil {
		if isMissingIngressControlTable(err) || isMissingFenceEpoch(err) {
			return Fence{}, schemaUnavailable(namespace, err)
		}
		return Fence{}, fmt.Errorf("connector admission: lock namespace fence %q: %w", namespace, err)
	}
	defer rows.Close()

	var fence *Fence
	for rows.Next() {
		var (
			scope      sql.NullString
			mode       string
			retryAfter int
			reason     string
			epoch      sql.NullInt64
			expiresAt  sql.NullTime
		)
		if err := rows.Scan(&scope, &mode, &retryAfter, &reason, &epoch, &expiresAt); err != nil {
			return Fence{}, fmt.Errorf("connector admission: scan namespace fence %q: %w", namespace, err)
		}
		if !scope.Valid || scope.String != "" || !epoch.Valid {
			return Fence{}, schemaUnavailable(namespace, fmt.Errorf("namespace fence has NULL scope or epoch"))
		}
		if fence != nil {
			return Fence{}, schemaUnavailable(namespace, fmt.Errorf("namespace fence has duplicate logical-global rows"))
		}
		fence = &Fence{
			Namespace:         namespace,
			Mode:              strings.ToLower(strings.TrimSpace(mode)),
			RetryAfterSeconds: retryAfter,
			Reason:            strings.TrimSpace(reason),
			Epoch:             epoch.Int64,
			Capability:        CapabilityAvailable,
		}
		if fence.Mode == "" {
			fence.Mode = ModeActive
		}
		if expiresAt.Valid && !expiresAt.Time.After(g.now().UTC()) {
			fence.Mode = ModeActive
		}
	}
	if err := rows.Err(); err != nil {
		return Fence{}, fmt.Errorf("connector admission: read namespace fence %q: %w", namespace, err)
	}
	if fence == nil {
		return Fence{}, schemaUnavailable(namespace, fmt.Errorf("namespace fence canonical row was not established"))
	}
	return *fence, nil
}

func requireOneCanonicalFenceUpdate(namespace string, result sql.Result) error {
	updated, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("connector admission: read namespace fence update result %q: %w", namespace, err)
	}
	if updated != 1 {
		return schemaUnavailable(namespace, fmt.Errorf("namespace fence canonical update affected %d rows", updated))
	}
	return nil
}

func Allow() Decision {
	return Decision{Allow: true, Mode: ModeActive}
}

func unavailableDecision() Decision {
	return Decision{Allow: false, Mode: ModeUnavailable, RetryAfterSeconds: 120, Reason: "ingress_control_schema_unavailable"}
}

// NamespaceFenceID is the deterministic primary key used only when the L0
// runtime must materialize a previously absent global fence row. Consumers
// should address the fence by its scoped tuple, not synthesize a second
// admission plan from this representation.
func NamespaceFenceID(namespace string) string {
	namespace = canonicalNamespace(namespace)
	sum := sha256.Sum256([]byte("nexuscrm.connector-ingress-fence/v1\x1f" + namespace + "\x1f\x1f*"))
	return "connector-ingress-fence-v1-" + fmt.Sprintf("%x", sum[:])
}

func canonicalNamespace(namespace string) string {
	return strings.ToLower(strings.TrimSpace(namespace))
}

func channelCandidates(channel string) []string {
	channel = strings.ToLower(strings.TrimSpace(channel))
	if channel == "" || channel == "*" {
		return []string{"*"}
	}
	parts := []string{channel}
	if prefix, _, ok := strings.Cut(channel, ":"); ok {
		prefix = strings.TrimSpace(prefix)
		if prefix != "" {
			parts = append(parts, prefix)
		}
	}
	parts = append(parts, "*")
	return parts
}

func placeholders(n int) string {
	if n <= 0 {
		return "?"
	}
	return strings.TrimRight(strings.Repeat("?,", n), ",")
}

func stringArgs(values []string) []interface{} {
	args := make([]interface{}, len(values))
	for i, value := range values {
		args[i] = value
	}
	return args
}

func isMissingIngressControlTable(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return (strings.Contains(msg, "_system_connectoringresscontrol") && strings.Contains(msg, "exist")) ||
		(strings.Contains(msg, "_system_connectoringresscontrol") && strings.Contains(msg, "unknown"))
}

func isMissingFenceEpoch(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "fenceepoch") && (strings.Contains(msg, "unknown column") || strings.Contains(msg, "doesn't exist"))
}

func schemaUnavailable(namespace string, cause error) error {
	return &SchemaUnavailableError{Namespace: canonicalNamespace(namespace), Cause: cause}
}