package persistence

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/base64"
	"errors"
	"fmt"
	"strings"
	"time"

	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

const RuntimePlatformCommandAccepted = "ACCEPTED"

var (
	ErrRuntimePlatformCommandConflict              = errors.New("runtime platform command idempotency identity is already bound to different content")
	ErrRuntimePlatformCommandServingBindingChanged = errors.New("runtime platform command is bound to a different serving revision")
	ErrRuntimePlatformCommandLegacyAuthority       = errors.New("runtime platform command contains pre-B2 rows without reconstructable execution authority")
)

type RuntimePlatformCommandRepository struct {
	db           *sql.DB
	transactions *TransactionManager
}

func NewRuntimePlatformCommandRepository(db *sql.DB, transactions ...*TransactionManager) *RuntimePlatformCommandRepository {
	var transactionManager *TransactionManager
	if len(transactions) == 1 {
		transactionManager = transactions[0]
	}
	return &RuntimePlatformCommandRepository{db: db, transactions: transactionManager}
}

var runtimePlatformCommandRequiredColumns = map[string]string{
	"Id":                              "varchar",
	"OrgId":                           "varchar",
	"AppInstallationId":               "varchar",
	"CallerWorkloadId":                "varchar",
	"RuntimeCapabilityId":             "varchar",
	"MetadataBinding":                 "mediumtext",
	"ExecutionScopeDigest":            "char",
	"ScopeDigest":                     "char",
	"CommandId":                       "varchar",
	"IdempotencyKey":                  "varchar",
	"CommandKind":                     "varchar",
	"PartitionKey":                    "varchar",
	"SerializationKeyDigest":          "char",
	"PayloadFormatVersion":            "bigint",
	"RequestDigest":                   "char",
	"Payload":                         "mediumblob",
	"Receipts":                        "json",
	"State":                           "varchar",
	"Attempt":                         "int",
	"AvailableAt":                     "datetime",
	"LeaseToken":                      "varchar",
	"LeaseExpiresAt":                  "datetime",
	"OwnershipEpoch":                  "decimal(20,0)",
	"DependencyKind":                  "varchar",
	"DependencyObjectAPIName":         "varchar",
	"DependencyRecordId":              "varchar",
	"DependencyKeyHash":               "char",
	"DependencyConnectorOwnerKey":     "varchar",
	"DependencyConnectorNamespace":    "varchar",
	"DependencyConnectorSourceObject": "varchar",
	"DependencyDestinationSlotId":     "varchar",
	"DependencyConnectorRemoteId":     "varchar",
	"DependencyBlockedAt":             "datetime",
	"LastError":                       "longtext",
	"RepairId":                        "varchar",
	"RepairReason":                    "varchar",
	"RepairCount":                     "int",
	"LastRepairAt":                    "datetime",
	"ResultPayload":                   "mediumtext",
	"PublicationIdentity":             "char",
	"PublishAttempt":                  "int",
	"RecoveryState":                   "varchar",
	"CreatedDate":                     "datetime",
	"LastModifiedDate":                "datetime",
}

const runtimePlatformCommandAuthorityInventoryQuery = `SELECT
Id, OrgId, AppInstallationId, CallerWorkloadId, RuntimeCapabilityId,
MetadataBinding, ExecutionScopeDigest
FROM _System_RuntimePlatformCommand
ORDER BY Id`

func VerifyRuntimePlatformCommandSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return errors.New("runtime platform command schema: context and database are required")
	}
	if err := VerifyRequiredTableColumns(
		ctx,
		db,
		"_System_RuntimePlatformCommand",
		runtimePlatformCommandRequiredColumns,
	); err != nil {
		return fmt.Errorf("runtime platform command schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		"_System_RuntimePlatformCommand",
		"ux_runtime_platform_command_scope",
		true,
		[]string{"ScopeDigest", "IdempotencyKey"},
	); err != nil {
		return fmt.Errorf("runtime platform command schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		"_System_RuntimePlatformCommand",
		"idx_runtime_platform_command_id",
		true,
		[]string{"ScopeDigest", "CommandId"},
	); err != nil {
		return fmt.Errorf("runtime platform command schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		"_System_RuntimePlatformCommand",
		"idx_runtime_platform_command_claim",
		false,
		[]string{"ExecutionScopeDigest", "State", "AvailableAt", "LeaseExpiresAt", "Id"},
	); err != nil {
		return fmt.Errorf("runtime platform command schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		"_System_RuntimePlatformCommand",
		"idx_runtime_platform_command_partition",
		false,
		[]string{"ExecutionScopeDigest", "PartitionKey", "State", "LeaseExpiresAt", "Id"},
	); err != nil {
		return fmt.Errorf("runtime platform command schema: %w", err)
	}
	if err := VerifyExactTableIndex(
		ctx,
		db,
		"_System_RuntimePlatformCommand",
		"idx_runtime_platform_command_serialization",
		false,
		[]string{"ExecutionScopeDigest", "SerializationKeyDigest", "State", "LeaseExpiresAt", "Id"},
	); err != nil {
		return fmt.Errorf("runtime platform command schema: %w", err)
	}
	legacyRows, err := countRuntimePlatformCommandRowsWithoutAuthority(ctx, db)
	if err != nil {
		return fmt.Errorf("runtime platform command schema: inspect stopped-cutover authority: %w", err)
	}
	if legacyRows != 0 {
		return fmt.Errorf(
			"%w: %d row(s) are quarantined in place; stop old command writers and explicitly drain or archive them before enabling CommandAcceptance or the Platform-command Worker",
			ErrRuntimePlatformCommandLegacyAuthority,
			legacyRows,
		)
	}
	return nil
}

func countRuntimePlatformCommandRowsWithoutAuthority(ctx context.Context, db *sql.DB) (int64, error) {
	rows, err := db.QueryContext(ctx, runtimePlatformCommandAuthorityInventoryQuery)
	if err != nil {
		return 0, err
	}
	defer rows.Close()

	var invalid int64
	for rows.Next() {
		var rowID, orgID, appInstallationID, callerWorkloadID sql.NullString
		var runtimeCapabilityID, metadataBinding, executionScopeDigest sql.NullString
		if err := rows.Scan(
			&rowID,
			&orgID,
			&appInstallationID,
			&callerWorkloadID,
			&runtimeCapabilityID,
			&metadataBinding,
			&executionScopeDigest,
		); err != nil {
			return 0, err
		}
		scope := RuntimeWorkScope{
			OrgID:               orgID.String,
			AppInstallationID:   appInstallationID.String,
			RuntimeCapabilityID: runtimeCapabilityID.String,
		}
		expectedDigest, digestErr := scope.Digest()
		if !rowID.Valid || rowID.String == "" ||
			!callerWorkloadID.Valid || callerWorkloadID.String == "" ||
			!metadataBinding.Valid || metadataBinding.String == "" ||
			!executionScopeDigest.Valid || digestErr != nil ||
			executionScopeDigest.String != expectedDigest {
			invalid++
		}
	}
	if err := rows.Err(); err != nil {
		return 0, err
	}
	return invalid, nil
}

type RuntimePlatformCommandAcceptance struct {
	RowID                string
	OrgID                string
	AppInstallationID    string
	CallerWorkloadID     string
	RuntimeCapabilityID  string
	MetadataBinding      string
	ExecutionScopeDigest string
	ScopeDigest          string
	CommandID            string
	IdempotencyKey       string
	CommandKind          string
	PartitionKey         string
	// SerializationKeyDigest is derived by trusted operation admission. It is
	// intentionally distinct from the caller-declared PartitionKey: PartitionKey
	// owns only source ordering, while this key owns aggregate single-flight.
	SerializationKeyDigest string
	PayloadFormatVersion   uint32
	RequestDigest          string
	Payload                []byte
	Receipts               []byte
}

type RuntimePlatformCommandRow struct {
	RowID                  string
	OrgID                  string
	AppInstallationID      string
	CallerWorkloadID       string
	RuntimeCapabilityID    string
	MetadataBinding        string
	ExecutionScopeDigest   string
	ScopeDigest            string
	CommandID              string
	IdempotencyKey         string
	CommandKind            string
	PartitionKey           string
	SerializationKeyDigest string
	PayloadFormatVersion   uint32
	RequestDigest          string
	Payload                []byte
	Receipts               []byte
	State                  string
	ResultPayload          []byte
	PublicationIdentity    string
	PublishAttempt         int
	RecoveryState          string
	Dependency             *contractmodels.FlowExecutionDependency
	DependencyBlockedAt    time.Time
}

// Accept atomically creates one durable, worker-visible command identity. It
// deliberately does not execute or route the command: the role-scoped Runtime
// Worker family owns that execution concern.
func (r *RuntimePlatformCommandRepository) Accept(ctx context.Context, input RuntimePlatformCommandAcceptance) (RuntimePlatformCommandRow, bool, error) {
	if r == nil || r.db == nil {
		return RuntimePlatformCommandRow{}, false, errors.New("runtime platform command repository is unavailable")
	}
	executionScopeDigest, scopeErr := (RuntimeWorkScope{
		OrgID:               input.OrgID,
		AppInstallationID:   input.AppInstallationID,
		RuntimeCapabilityID: input.RuntimeCapabilityID,
	}).Digest()
	if !hexDigestPattern.MatchString(input.ScopeDigest) ||
		(input.SerializationKeyDigest != "" && !hexDigestPattern.MatchString(input.SerializationKeyDigest)) || input.RowID == "" ||
		scopeErr != nil || input.CallerWorkloadID == "" ||
		strings.TrimSpace(input.CallerWorkloadID) != input.CallerWorkloadID ||
		len(input.CallerWorkloadID) > 1024 || input.MetadataBinding == "" ||
		input.ExecutionScopeDigest != executionScopeDigest || input.CommandID == "" ||
		input.IdempotencyKey == "" || input.CommandKind == "" || input.RequestDigest == "" ||
		input.PayloadFormatVersion == 0 || len(input.Payload) == 0 || len(input.Receipts) == 0 {
		return RuntimePlatformCommandRow{}, false, errors.New("runtime platform command acceptance is incomplete")
	}
	exec := r.executor(ctx)
	result, err := exec.ExecContext(ctx, `INSERT IGNORE INTO _System_RuntimePlatformCommand
(Id, OrgId, AppInstallationId, CallerWorkloadId, RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest,
 ScopeDigest, CommandId, IdempotencyKey, CommandKind, PartitionKey, SerializationKeyDigest, PayloadFormatVersion, RequestDigest, Payload, Receipts,
 State, AvailableAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'ACCEPTED',
        UTC_TIMESTAMP(6), UTC_TIMESTAMP(6), UTC_TIMESTAMP(6))`,
		input.RowID, input.OrgID, input.AppInstallationID, input.CallerWorkloadID,
		input.RuntimeCapabilityID, input.MetadataBinding, input.ExecutionScopeDigest,
		input.ScopeDigest, input.CommandID, input.IdempotencyKey, input.CommandKind,
		input.PartitionKey, nullableString(input.SerializationKeyDigest), input.PayloadFormatVersion, input.RequestDigest, input.Payload, input.Receipts,
	)
	if err != nil {
		return RuntimePlatformCommandRow{}, false, fmt.Errorf("accept runtime platform command: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return RuntimePlatformCommandRow{}, false, fmt.Errorf("accept runtime platform command rows affected: %w", err)
	}
	row, err := r.load(ctx, exec, input.ScopeDigest, input.CommandID, input.IdempotencyKey)
	if err != nil {
		return RuntimePlatformCommandRow{}, false, err
	}
	if row.CommandID != input.CommandID || row.IdempotencyKey != input.IdempotencyKey ||
		row.OrgID != input.OrgID || row.AppInstallationID != input.AppInstallationID ||
		row.CallerWorkloadID != input.CallerWorkloadID ||
		row.RuntimeCapabilityID != input.RuntimeCapabilityID ||
		row.ExecutionScopeDigest != input.ExecutionScopeDigest {
		return row, true, ErrRuntimePlatformCommandConflict
	}
	if row.MetadataBinding != input.MetadataBinding {
		return row, true, ErrRuntimePlatformCommandServingBindingChanged
	}
	sameReceipts, err := equivalentJSONDocuments(row.Receipts, input.Receipts)
	if err != nil {
		return row, true, fmt.Errorf("compare runtime platform command receipts: %w", err)
	}
	if !sameReceipts {
		return row, true, ErrRuntimePlatformCommandServingBindingChanged
	}
	if row.CommandKind != input.CommandKind ||
		row.PartitionKey != input.PartitionKey || row.SerializationKeyDigest != input.SerializationKeyDigest ||
		row.PayloadFormatVersion != input.PayloadFormatVersion ||
		row.RequestDigest != input.RequestDigest {
		return row, true, ErrRuntimePlatformCommandConflict
	}
	if !bytes.Equal(row.Payload, input.Payload) {
		return row, true, ErrRuntimePlatformCommandConflict
	}
	return row, affected == 0, nil
}

func (r *RuntimePlatformCommandRepository) load(ctx context.Context, exec Executor, scopeDigest, commandID, idempotencyKey string) (RuntimePlatformCommandRow, error) {
	rows, err := exec.QueryContext(ctx, `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest, CommandId, IdempotencyKey, CommandKind,
       COALESCE(PartitionKey, ''), COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion, RequestDigest, Payload, Receipts, State, COALESCE(ResultPayload, '')
  FROM _System_RuntimePlatformCommand
 WHERE ScopeDigest = ? AND (CommandId = ? OR IdempotencyKey = ?)
 ORDER BY Id
 LIMIT 2`, scopeDigest, commandID, idempotencyKey)
	if err != nil {
		return RuntimePlatformCommandRow{}, fmt.Errorf("load runtime platform command: %w", err)
	}
	defer rows.Close()
	var result RuntimePlatformCommandRow
	count := 0
	for rows.Next() {
		var row RuntimePlatformCommandRow
		var encodedResult string
		if err := rows.Scan(
			&row.RowID, &row.OrgID, &row.AppInstallationID, &row.CallerWorkloadID,
			&row.RuntimeCapabilityID, &row.MetadataBinding, &row.ExecutionScopeDigest,
			&row.ScopeDigest, &row.CommandID, &row.IdempotencyKey, &row.CommandKind,
			&row.PartitionKey, &row.SerializationKeyDigest, &row.PayloadFormatVersion, &row.RequestDigest, &row.Payload, &row.Receipts, &row.State,
			&encodedResult,
		); err != nil {
			return RuntimePlatformCommandRow{}, fmt.Errorf("load runtime platform command row: %w", err)
		}
		row.ResultPayload, err = decodeRuntimePlatformCommandResult(encodedResult)
		if err != nil {
			return RuntimePlatformCommandRow{}, fmt.Errorf("load runtime platform command result: %w", err)
		}
		count++
		result = row
	}
	if err := rows.Err(); err != nil {
		return RuntimePlatformCommandRow{}, fmt.Errorf("load runtime platform command rows: %w", err)
	}
	if count == 0 {
		return RuntimePlatformCommandRow{}, errors.New("runtime platform command insert was not observable")
	}
	if count != 1 {
		return RuntimePlatformCommandRow{}, ErrRuntimePlatformCommandConflict
	}
	return result, nil
}

// Get returns one command only from the complete host-authenticated acceptance
// partition and only when the caller presents the original request digest.
// It is the neutral status read used by typed operation owners; it does not
// interpret an operation-specific result payload.
func (r *RuntimePlatformCommandRepository) Get(
	ctx context.Context,
	scope RuntimePlatformCommandTrustedScope,
	commandID string,
	requestDigest string,
) (RuntimePlatformCommandRow, bool, error) {
	if commandID == "" {
		return RuntimePlatformCommandRow{}, false, errors.New("runtime platform command status request is incomplete")
	}
	return r.LookupAcceptance(ctx, scope, commandID, "", requestDigest)
}

// LookupAcceptance resolves either caller-authored durable identity in the
// complete host-authenticated acceptance partition. Typed operation admission
// uses it before consulting mutable metadata so a replay cannot become
// dependent on a later Catalog view. Presenting the other identity or request
// content for an existing slot is a conflict, never a second acceptance.
func (r *RuntimePlatformCommandRepository) LookupAcceptance(
	ctx context.Context,
	scope RuntimePlatformCommandTrustedScope,
	commandID string,
	idempotencyKey string,
	requestDigest string,
) (RuntimePlatformCommandRow, bool, error) {
	if r == nil || r.db == nil || (commandID == "" && idempotencyKey == "") || !validPrefixedSHA256(requestDigest) {
		return RuntimePlatformCommandRow{}, false, errors.New("runtime platform command acceptance lookup is incomplete")
	}
	scopeDigest, err := scope.Digest()
	if err != nil {
		return RuntimePlatformCommandRow{}, false, err
	}
	rows, err := r.executor(ctx).QueryContext(ctx, `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest, CommandId, IdempotencyKey,
       CommandKind, COALESCE(PartitionKey, ''), COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion,
       RequestDigest, Payload, Receipts, State, ResultPayload,
       COALESCE(PublicationIdentity, ''), PublishAttempt, COALESCE(RecoveryState, '')
FROM _System_RuntimePlatformCommand
WHERE ScopeDigest = ? AND (CommandId = ? OR IdempotencyKey = ?)
  AND OrgId = ? AND AppInstallationId = ? AND CallerWorkloadId = ? AND RuntimeCapabilityId = ?
ORDER BY Id
LIMIT 2`,
		scopeDigest, commandID, idempotencyKey, scope.OrgID, scope.AppInstallationID,
		scope.CallerWorkloadID, scope.RuntimeCapabilityID,
	)
	if err != nil {
		return RuntimePlatformCommandRow{}, false, fmt.Errorf("look up runtime platform command acceptance: %w", err)
	}
	defer rows.Close()
	var row RuntimePlatformCommandRow
	count := 0
	for rows.Next() {
		var candidate RuntimePlatformCommandRow
		var encodedResult sql.NullString
		if err := rows.Scan(
			&candidate.RowID, &candidate.OrgID, &candidate.AppInstallationID, &candidate.CallerWorkloadID,
			&candidate.RuntimeCapabilityID, &candidate.MetadataBinding, &candidate.ExecutionScopeDigest, &candidate.ScopeDigest,
			&candidate.CommandID, &candidate.IdempotencyKey, &candidate.CommandKind, &candidate.PartitionKey,
			&candidate.SerializationKeyDigest, &candidate.PayloadFormatVersion, &candidate.RequestDigest,
			&candidate.Payload, &candidate.Receipts, &candidate.State, &encodedResult,
			&candidate.PublicationIdentity, &candidate.PublishAttempt, &candidate.RecoveryState,
		); err != nil {
			return RuntimePlatformCommandRow{}, false, fmt.Errorf("scan runtime platform command acceptance: %w", err)
		}
		candidate.ResultPayload, err = decodeNullableRuntimePlatformCommandResult(encodedResult)
		if err != nil {
			return RuntimePlatformCommandRow{}, true, fmt.Errorf("decode runtime platform command status result: %w", err)
		}
		row = candidate
		count++
	}
	if err := rows.Err(); err != nil {
		return RuntimePlatformCommandRow{}, false, fmt.Errorf("read runtime platform command acceptance rows: %w", err)
	}
	if count == 0 {
		return RuntimePlatformCommandRow{}, false, nil
	}
	if count != 1 {
		return RuntimePlatformCommandRow{}, true, ErrRuntimePlatformCommandConflict
	}
	if row.CommandID != commandID || (idempotencyKey != "" && row.IdempotencyKey != idempotencyKey) ||
		row.RequestDigest != requestDigest {
		return RuntimePlatformCommandRow{}, true, ErrRuntimePlatformCommandConflict
	}
	return row, true, nil
}

const runtimePlatformCommandResultPrefix = "base64:"

func encodeRuntimePlatformCommandResult(value []byte) string {
	return runtimePlatformCommandResultPrefix + base64.StdEncoding.EncodeToString(value)
}

func decodeRuntimePlatformCommandResult(value string) ([]byte, error) {
	if value == "" {
		return nil, nil
	}
	if !strings.HasPrefix(value, runtimePlatformCommandResultPrefix) {
		return nil, errors.New("runtime platform command terminal result encoding is invalid")
	}
	decoded, err := base64.StdEncoding.Strict().DecodeString(strings.TrimPrefix(value, runtimePlatformCommandResultPrefix))
	if err != nil || len(decoded) == 0 {
		return nil, errors.New("runtime platform command terminal result encoding is invalid")
	}
	return decoded, nil
}

func (r *RuntimePlatformCommandRepository) executor(ctx context.Context) Executor {
	if r != nil && r.transactions != nil {
		if tx, ok := r.transactions.ExtractTx(ctx).(*sql.Tx); ok && tx != nil {
			return tx
		}
	}
	return r.db
}