package runtimehandoff

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"strings"

	connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
)

var (
	ErrLeaseLost = errors.New(
		"connector Runtime command handoff lost its WorkItem lease",
	)
	ErrConflict = errors.New(
		"connector Runtime command handoff is already bound to different content",
	)
)

// storeUnavailableError marks an Integration-store operation that can be
// retried without changing the frozen command identity.
type storeUnavailableError struct {
	Cause error
}

func (e *storeUnavailableError) Error() string {
	if e == nil || e.Cause == nil {
		return "connector Runtime command handoff store is unavailable"
	}
	return e.Cause.Error()
}

func (e *storeUnavailableError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Cause
}

// PreparedCommand is the immutable Integration-owned side of one private
// Runtime command handoff. The Runtime owns acceptance and execution after the
// request succeeds; this row only freezes retry bytes across an uncertain gRPC
// outcome or Integration restart.
type PreparedCommand struct {
	OrgID                  string
	AppInstallationID      string
	AuthorizedWorkloadID   string
	ScopeDigest            string
	MinimumCatalogRevision uint64
	BindingViewDigest      string
	Request                commandv1.Request
}

type Store struct {
	db *sql.DB
}

func NewStore(db *sql.DB) (*Store, error) {
	if db == nil {
		return nil, errors.New(
			"connector Runtime command handoff store requires a database",
		)
	}
	return &Store{db: db}, nil
}

// VerifySchema fails closed before a Worker begins claiming work when the
// Controlplane-published immutable receipt schema is absent or incomplete.
func (s *Store) VerifySchema(ctx context.Context) error {
	if s == nil || s.db == nil {
		return errors.New(
			"connector Runtime command handoff store is unavailable",
		)
	}
	rows, err := s.db.QueryContext(ctx, `
SELECT WorkId, EventId, Namespace, InstalledPackId, EventKind, RouteVersion,
       OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest,
       MinimumCatalogRevision, BindingViewDigest, CommandId, IdempotencyKey,
       CommandKind, PartitionKey, PayloadFormatVersion, Payload, PayloadDigest,
       CreatedDate, LastModifiedDate
FROM _System_ConnectorRuntimeHandoffReceipt
WHERE 1 = 0`)
	if err != nil {
		return fmt.Errorf(
			"verify connector Runtime handoff receipt schema: %w",
			err,
		)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf(
			"verify connector Runtime handoff receipt columns: %w",
			err,
		)
	}
	return s.verifyReceiptIndexes(ctx)
}

func (s *Store) verifyReceiptIndexes(ctx context.Context) error {
	rows, err := s.db.QueryContext(ctx, `
SELECT INDEX_NAME, NON_UNIQUE, SEQ_IN_INDEX, COLUMN_NAME, SUB_PART
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = '_System_ConnectorRuntimeHandoffReceipt'
  AND INDEX_NAME IN (
    'PRIMARY',
    'ux_connector_runtime_handoff_command',
    'ux_connector_runtime_handoff_idempotency'
  )
ORDER BY INDEX_NAME, SEQ_IN_INDEX`)
	if err != nil {
		return fmt.Errorf(
			"verify connector Runtime handoff receipt indexes: %w",
			err,
		)
	}
	defer rows.Close()

	type indexShape struct {
		unique  bool
		columns []string
	}
	actual := make(map[string]indexShape, 3)
	for rows.Next() {
		var (
			name      string
			nonUnique int
			sequence  int
			column    string
			subPart   sql.NullInt64
		)
		if err := rows.Scan(
			&name,
			&nonUnique,
			&sequence,
			&column,
			&subPart,
		); err != nil {
			return fmt.Errorf(
				"read connector Runtime handoff receipt index: %w",
				err,
			)
		}
		shape := actual[name]
		if nonUnique != 0 || sequence != len(shape.columns)+1 || subPart.Valid {
			return errors.New(
				"connector Runtime handoff receipt index shape is unsafe",
			)
		}
		shape.unique = true
		shape.columns = append(shape.columns, column)
		actual[name] = shape
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf(
			"read connector Runtime handoff receipt indexes: %w",
			err,
		)
	}
	expected := map[string][]string{
		"PRIMARY": {"WorkId"},
		"ux_connector_runtime_handoff_command": {
			"ScopeDigest", "CommandId",
		},
		"ux_connector_runtime_handoff_idempotency": {
			"ScopeDigest", "IdempotencyKey",
		},
	}
	for name, columns := range expected {
		shape, found := actual[name]
		if !found || !shape.unique || len(shape.columns) != len(columns) {
			return errors.New(
				"connector Runtime handoff receipt uniqueness is unavailable",
			)
		}
		for index := range columns {
			if shape.columns[index] != columns[index] {
				return errors.New(
					"connector Runtime handoff receipt uniqueness is unavailable",
				)
			}
		}
	}
	return nil
}

// Prepare inserts request bytes only while the exact WorkItem lease is still
// live. A retry always returns the first committed bytes; it never rebuilds a
// request from a newer provider read after an uncertain Runtime outcome.
func (s *Store) Prepare(
	ctx context.Context,
	scope connector_inbound.RuntimeCommandHandoffScope,
	candidate PreparedCommand,
) (PreparedCommand, bool, error) {
	if s == nil || s.db == nil {
		return PreparedCommand{}, false, errors.New(
			"connector Runtime command handoff store is unavailable",
		)
	}
	canonicalScope, err := connector_inbound.CanonicalRuntimeCommandHandoffScope(scope)
	if err != nil {
		return PreparedCommand{}, false, err
	}
	scope = canonicalScope
	if err := validatePreparedCommand(ctx, scope, candidate); err != nil {
		return PreparedCommand{}, false, err
	}
	request := candidate.Request
	result, err := s.db.ExecContext(ctx, `
INSERT IGNORE INTO _System_ConnectorRuntimeHandoffReceipt
       (WorkId, EventId, Namespace, InstalledPackId, EventKind, RouteVersion,
        OrgId, AppInstallationId, AuthorizedWorkloadId,
        ScopeDigest, MinimumCatalogRevision, BindingViewDigest, CommandId, IdempotencyKey,
        CommandKind, PartitionKey, PayloadFormatVersion, Payload,
        PayloadDigest, CreatedDate)
SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?, UTC_TIMESTAMP(6)
FROM _System_ConnectorWorkItem
WHERE WorkId = ? AND EventId = ? AND State = 'processing' AND LeaseToken = ?
  AND LeaseUntil > UTC_TIMESTAMP(6)
LIMIT 1`,
		scope.WorkItemID,
		scope.EventID,
		scope.Namespace,
		scope.InstalledPackID,
		scope.EventKind,
		strconv.FormatUint(scope.RouteVersion, 10),
		candidate.OrgID,
		candidate.AppInstallationID,
		candidate.AuthorizedWorkloadID,
		candidate.ScopeDigest,
		strconv.FormatUint(candidate.MinimumCatalogRevision, 10),
		candidate.BindingViewDigest,
		request.CommandID,
		request.IdempotencyKey,
		request.CommandKind,
		request.PartitionKey,
		request.PayloadFormatVersion,
		request.Payload,
		request.PayloadDigest,
		scope.WorkItemID,
		scope.EventID,
		scope.LeaseToken,
	)
	if err != nil {
		return PreparedCommand{}, false, &storeUnavailableError{Cause: fmt.Errorf(
			"prepare connector Runtime command handoff: %w", err,
		)}
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return PreparedCommand{}, false, &storeUnavailableError{Cause: fmt.Errorf(
			"observe connector Runtime command handoff insert: %w", err,
		)}
	}
	stored, found, err := s.load(ctx, scope)
	if err != nil {
		return PreparedCommand{}, false, err
	}
	if !found {
		return PreparedCommand{}, false, ErrLeaseLost
	}
	if !samePreparedCommand(stored, candidate) {
		return PreparedCommand{}, affected == 0, ErrConflict
	}
	return stored, affected == 0, nil
}

// LoadForScope returns previously frozen bytes before Integration repeats
// provider enrichment. This is the uncertain-outcome restart path: once a
// request was prepared, newer external data must not silently create different
// command content for the same WorkItem identity.
func (s *Store) LoadForScope(
	ctx context.Context,
	scope connector_inbound.RuntimeCommandHandoffScope,
) (PreparedCommand, bool, error) {
	if s == nil || s.db == nil {
		return PreparedCommand{}, false, errors.New(
			"connector Runtime command handoff store is unavailable",
		)
	}
	canonicalScope, err := connector_inbound.CanonicalRuntimeCommandHandoffScope(scope)
	if err != nil {
		return PreparedCommand{}, false, err
	}
	return s.load(ctx, canonicalScope)
}

func (s *Store) load(
	ctx context.Context,
	scope connector_inbound.RuntimeCommandHandoffScope,
) (PreparedCommand, bool, error) {
	var (
		stored          PreparedCommand
		storedNamespace string
		storedPackID    string
		storedEventKind string
		rawRouteVersion []byte
		rawRevision     []byte
		payloadVersion  uint64
		partitionKey    sql.NullString
		leaseLive       bool
	)
	err := s.db.QueryRowContext(ctx, `
SELECT r.Namespace, r.InstalledPackId, r.EventKind, r.RouteVersion,
       r.OrgId, r.AppInstallationId, r.AuthorizedWorkloadId, r.ScopeDigest,
       r.MinimumCatalogRevision, r.BindingViewDigest, r.CommandId,
       r.IdempotencyKey, r.CommandKind, r.PartitionKey,
       r.PayloadFormatVersion, r.Payload, r.PayloadDigest,
       EXISTS (
         SELECT 1
         FROM _System_ConnectorWorkItem w
         WHERE w.WorkId = r.WorkId AND w.EventId = r.EventId
           AND w.State = 'processing' AND w.LeaseToken = ?
           AND w.LeaseUntil > UTC_TIMESTAMP(6)
       ) AS LeaseLive
FROM _System_ConnectorRuntimeHandoffReceipt r
WHERE r.WorkId = ? AND r.EventId = ?
LIMIT 1`,
		scope.LeaseToken,
		scope.WorkItemID,
		scope.EventID,
	).Scan(
		&storedNamespace,
		&storedPackID,
		&storedEventKind,
		&rawRouteVersion,
		&stored.OrgID,
		&stored.AppInstallationID,
		&stored.AuthorizedWorkloadID,
		&stored.ScopeDigest,
		&rawRevision,
		&stored.BindingViewDigest,
		&stored.Request.CommandID,
		&stored.Request.IdempotencyKey,
		&stored.Request.CommandKind,
		&partitionKey,
		&payloadVersion,
		&stored.Request.Payload,
		&stored.Request.PayloadDigest,
		&leaseLive,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return PreparedCommand{}, false, nil
	}
	if err != nil {
		return PreparedCommand{}, false, &storeUnavailableError{Cause: fmt.Errorf(
			"load connector Runtime command handoff: %w", err,
		)}
	}
	if !leaseLive {
		return PreparedCommand{}, false, ErrLeaseLost
	}
	routeVersion, routeErr := strconv.ParseUint(string(rawRouteVersion), 10, 64)
	revision, revisionErr := strconv.ParseUint(string(rawRevision), 10, 64)
	if routeErr != nil || revisionErr != nil || routeVersion == 0 || revision == 0 ||
		storedNamespace != scope.Namespace || storedPackID != scope.InstalledPackID ||
		storedEventKind != scope.EventKind || routeVersion != scope.RouteVersion ||
		payloadVersion == 0 ||
		payloadVersion > uint64(^uint32(0)) {
		return PreparedCommand{}, false, errors.New(
			"connector Runtime command handoff contains invalid numeric authority",
		)
	}
	stored.MinimumCatalogRevision = revision
	stored.Request.PartitionKey = partitionKey.String
	stored.Request.PayloadFormatVersion = uint32(payloadVersion)
	stored.Request.MinCatalogRevision = &stored.MinimumCatalogRevision
	if err := validatePreparedCommand(
		ctx,
		scope,
		stored,
	); err != nil {
		return PreparedCommand{}, false, errors.New(
			"connector Runtime command handoff contains invalid stored authority",
		)
	}
	return stored, true, nil
}

func validatePreparedCommand(
	ctx context.Context,
	scope connector_inbound.RuntimeCommandHandoffScope,
	value PreparedCommand,
) error {
	if scope.WorkItemID == "" || scope.EventID == "" ||
		scope.LeaseToken == "" || scope.Namespace == "" ||
		scope.InstalledPackID == "" || scope.EventKind == "" ||
		strings.TrimSpace(value.OrgID) == "" ||
		strings.TrimSpace(value.AppInstallationID) == "" ||
		strings.TrimSpace(value.AuthorizedWorkloadID) == "" ||
		!validHexDigest(value.ScopeDigest) ||
		value.MinimumCatalogRevision == 0 ||
		!validSHA256Digest(value.BindingViewDigest) ||
		value.Request.MinCatalogRevision == nil ||
		*value.Request.MinCatalogRevision != value.MinimumCatalogRevision {
		return errors.New(
			"connector Runtime command handoff authority is incomplete",
		)
	}
	if err := commandv1.ValidateRequest(ctx, value.Request); err != nil {
		return fmt.Errorf(
			"connector Runtime command handoff request is invalid: %w",
			err,
		)
	}
	return nil
}

func validSHA256Digest(value string) bool {
	if len(value) != len("sha256:")+64 || !strings.HasPrefix(value, "sha256:") {
		return false
	}
	decoded, err := hex.DecodeString(strings.TrimPrefix(value, "sha256:"))
	return err == nil && len(decoded) == sha256.Size
}

func samePreparedCommand(left, right PreparedCommand) bool {
	if left.OrgID != right.OrgID ||
		left.AppInstallationID != right.AppInstallationID ||
		left.AuthorizedWorkloadID != right.AuthorizedWorkloadID ||
		left.ScopeDigest != right.ScopeDigest ||
		left.MinimumCatalogRevision != right.MinimumCatalogRevision ||
		left.BindingViewDigest != right.BindingViewDigest {
		return false
	}
	leftRequest := left.Request
	rightRequest := right.Request
	return leftRequest.CommandID == rightRequest.CommandID &&
		leftRequest.IdempotencyKey == rightRequest.IdempotencyKey &&
		leftRequest.CommandKind == rightRequest.CommandKind &&
		leftRequest.PartitionKey == rightRequest.PartitionKey &&
		leftRequest.PayloadFormatVersion == rightRequest.PayloadFormatVersion &&
		leftRequest.PayloadDigest == rightRequest.PayloadDigest &&
		bytes.Equal(leftRequest.Payload, rightRequest.Payload)
}

func validHexDigest(value string) bool {
	if len(value) != sha256.Size*2 {
		return false
	}
	decoded, err := hex.DecodeString(value)
	return err == nil && len(decoded) == sha256.Size
}