package persistence

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"
)

const runtimePlatformCommandPublicationIdentityPrefix = "sha256:"

// RuntimePlatformCommandPublicationLease is deliberately distinct from an
// execution lease. Its Attempt counts only Catalog publication/reconciliation
// work; the command's business execution Attempt is immutable in this phase.
type RuntimePlatformCommandPublicationLease struct {
	Row               RuntimePlatformCommandRow
	LeaseToken        string
	Attempt           int
	AttemptsRemaining int
	LeaseExpiresAt    time.Time
	OwnershipEpoch    uint64
}

// PreparePublication commits the Runtime effect, opaque pending result, and
// deterministic publication identity in the same transaction. Once this
// method commits, the command can only be reclaimed through ClaimPublication;
// its operation executor must never be invoked again.
func (s *RuntimePlatformCommandExecutionStore) PreparePublication(
	ctx context.Context,
	lease RuntimePlatformCommandLease,
	effect func(context.Context) ([]byte, error),
) error {
	if err := s.validateLease(lease); err != nil {
		return err
	}
	if effect == nil {
		return errors.New("runtime platform command publication: effect callback is required")
	}
	completion, err := s.executeTransition(
		ctx, lease, false, RuntimePlatformCommandPublishing, effect,
	)
	return runtimePlatformCommandTransactionResult(completion, err)
}

// HasPublicationWork is a negative-only hint. ClaimPublication remains the
// fenced authority and also converts an exhausted publication phase to a
// phase-aware poison record.
func (s *RuntimePlatformCommandExecutionStore) HasPublicationWork(ctx context.Context) (bool, error) {
	if s == nil || s.db == nil {
		return false, errors.New("runtime platform command publication: store is unavailable")
	}
	var exists bool
	if err := s.db.QueryRowContext(ctx, `SELECT EXISTS(
  SELECT 1
  FROM _System_RuntimePlatformCommand USE INDEX (idx_runtime_platform_command_claim)
  WHERE ExecutionScopeDigest = ?
    AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
    AND State = 'PUBLISHING' AND AvailableAt <= UTC_TIMESTAMP(6)
    AND (LeaseToken IS NULL OR LeaseExpiresAt <= UTC_TIMESTAMP(6))
  LIMIT 1
)`, s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID).Scan(&exists); err != nil {
		return false, fmt.Errorf("runtime platform command publication: hint: %w", err)
	}
	return exists, nil
}

// ClaimPublication leases only post-commit publication/reconciliation work.
// It never changes the business execution Attempt and never makes the row
// eligible for Claim, which only owns pre-effect execution states.
func (s *RuntimePlatformCommandExecutionStore) ClaimPublication(
	ctx context.Context,
) (RuntimePlatformCommandPublicationLease, bool, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return RuntimePlatformCommandPublicationLease{}, false,
			errors.New("runtime platform command publication: store is unavailable")
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return RuntimePlatformCommandPublicationLease{}, false,
			fmt.Errorf("runtime platform command publication: begin claim: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (RuntimePlatformCommandPublicationLease, bool, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return RuntimePlatformCommandPublicationLease{}, false, errors.Join(cause, rollbackErr)
		}
		return RuntimePlatformCommandPublicationLease{}, false, cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var databaseNow time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&databaseNow); err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: read claim time: %w", err))
	}
	var row RuntimePlatformCommandRow
	var encodedPending sql.NullString
	var businessAttempt int
	err = tx.QueryRowContext(ctx, `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest, CommandId, IdempotencyKey,
       CommandKind, COALESCE(PartitionKey, ''), COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion, RequestDigest,
       Payload, Receipts, State, Attempt, COALESCE(PublicationIdentity, ''), PublishAttempt,
       ResultPayload
FROM _System_RuntimePlatformCommand USE INDEX (idx_runtime_platform_command_claim)
WHERE ExecutionScopeDigest = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PUBLISHING' AND AvailableAt <= ?
  AND (LeaseToken IS NULL OR LeaseExpiresAt <= ?)
ORDER BY AvailableAt, Id
LIMIT 1 FOR UPDATE`,
		s.scopeDigest, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		databaseNow, databaseNow,
	).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,
		&businessAttempt, &row.PublicationIdentity, &row.PublishAttempt, &encodedPending,
	)
	if errors.Is(err, sql.ErrNoRows) {
		if err := tx.Commit(); err != nil {
			return RuntimePlatformCommandPublicationLease{}, false,
				fmt.Errorf("runtime platform command publication: empty claim commit: %w", err)
		}
		return RuntimePlatformCommandPublicationLease{}, false, nil
	}
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: select claim: %w", err))
	}
	row.State = RuntimePlatformCommandPublishing
	row.ResultPayload, err = decodeNullableRuntimePlatformCommandResult(encodedPending)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: decode pending result: %w", err))
	}
	expectedIdentity, err := runtimePlatformCommandPublicationIdentity(
		row.ScopeDigest, row.CommandID, row.RequestDigest,
	)
	if err != nil || expectedIdentity != row.PublicationIdentity || businessAttempt <= 0 || len(row.ResultPayload) == 0 ||
		row.ExecutionScopeDigest != s.scopeDigest || row.OrgID != s.scope.OrgID ||
		row.AppInstallationID != s.scope.AppInstallationID || row.RuntimeCapabilityID != s.scope.RuntimeCapabilityID {
		return rollback(errors.New("runtime platform command publication: selected row authority is invalid"))
	}
	if row.PublishAttempt >= s.maxPublicationAttempts {
		result, updateErr := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'POISONED', RecoveryState = 'PUBLISHING',
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PUBLISHING' AND PublishAttempt = ?`,
			"bounded publication attempts exhausted", row.RowID, row.PublishAttempt,
		)
		if updateErr != nil {
			return rollback(fmt.Errorf("runtime platform command publication: poison exhausted row: %w", updateErr))
		}
		if affected, updateErr := result.RowsAffected(); updateErr != nil || affected != 1 {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		if err := tx.Commit(); err != nil {
			return RuntimePlatformCommandPublicationLease{}, false,
				fmt.Errorf("runtime platform command publication: poison commit outcome unknown: %w", err)
		}
		return RuntimePlatformCommandPublicationLease{}, false, nil
	}
	token := s.tokenFn()
	nextAttempt := row.PublishAttempt + 1
	databaseExpiry := databaseNow.Add(s.leaseDuration)
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET PublishAttempt = ?, LeaseToken = ?, LeaseExpiresAt = ?, OwnershipEpoch = ?,
    LastError = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PUBLISHING' AND PublishAttempt = ?
  AND (LeaseToken IS NULL OR LeaseExpiresAt <= ?)`,
		nextAttempt, token, databaseExpiry, s.expected.EpochDecimal(), row.RowID,
		row.PublishAttempt, databaseNow,
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: claim row: %w", err))
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return RuntimePlatformCommandPublicationLease{}, false,
			fmt.Errorf("runtime platform command publication: claim commit outcome unknown: %w", err)
	}
	row.PublishAttempt = nextAttempt
	return RuntimePlatformCommandPublicationLease{
		Row: row, LeaseToken: token, Attempt: nextAttempt,
		AttemptsRemaining: s.maxPublicationAttempts - nextAttempt,
		LeaseExpiresAt:    runtimePlatformCommandDatabaseUTC(databaseExpiry),
		OwnershipEpoch:    s.expected.OwnershipEpoch,
	}, true, nil
}

// RetryPublication releases only the publication lease. It cannot return the
// command to ACCEPTED or change the business execution Attempt. Exhaustion is
// repairable POISONED/PUBLISHING, never a terminal business failure.
func (s *RuntimePlatformCommandExecutionStore) RetryPublication(
	ctx context.Context,
	lease RuntimePlatformCommandPublicationLease,
	cause error,
	backoff time.Duration,
) error {
	if err := s.validatePublicationLease(lease); err != nil {
		return err
	}
	if cause == nil || backoff < 0 || backoff > 5*time.Minute {
		return errors.New("runtime platform command publication: retry is incomplete")
	}
	message := strings.TrimSpace(cause.Error())
	if message == "" {
		message = "publication failed"
	}
	if len(message) > 4096 {
		message = message[:4096]
	}
	return s.releasePublicationLease(ctx, lease, message, backoff)
}

// CompletePublication atomically acknowledges one exact publication lease,
// replaces the opaque pending postcondition with its verified terminal result,
// and advances the serving receipts that produced that result. It does not run
// a business-effect callback: a command in PUBLISHING has already committed its
// mutation and may only be completed through this fenced acknowledgement.
func (s *RuntimePlatformCommandExecutionStore) CompletePublication(
	ctx context.Context,
	lease RuntimePlatformCommandPublicationLease,
	terminalResult []byte,
	receipts []byte,
) error {
	if err := s.validatePublicationLease(lease); err != nil {
		return err
	}
	if len(terminalResult) == 0 || len(terminalResult) > s.publicationTerminalResultMaxBytes ||
		len(receipts) == 0 || !json.Valid(receipts) {
		return ErrRuntimePlatformCommandResultInvalid
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command publication: begin completion: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) error {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return errors.Join(cause, rollbackErr)
		}
		return cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var locked RuntimePlatformCommandRow
	var encodedPending sql.NullString
	var state, token, epoch string
	var businessAttempt, publishAttempt int
	var expiresAt time.Time
	if err := tx.QueryRowContext(ctx, `SELECT Id, OrgId, AppInstallationId, CallerWorkloadId,
       RuntimeCapabilityId, MetadataBinding, ExecutionScopeDigest, ScopeDigest,
       CommandId, IdempotencyKey, CommandKind, COALESCE(PartitionKey, ''),
       COALESCE(SerializationKeyDigest, ''), PayloadFormatVersion, RequestDigest,
       Payload, Receipts, State, COALESCE(LeaseToken, ''), Attempt,
       COALESCE(CAST(OwnershipEpoch AS CHAR), '0'), LeaseExpiresAt,
       COALESCE(PublicationIdentity, ''), PublishAttempt, ResultPayload
FROM _System_RuntimePlatformCommand
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`,
		lease.Row.RowID, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
	).Scan(
		&locked.RowID, &locked.OrgID, &locked.AppInstallationID, &locked.CallerWorkloadID,
		&locked.RuntimeCapabilityID, &locked.MetadataBinding, &locked.ExecutionScopeDigest,
		&locked.ScopeDigest, &locked.CommandID, &locked.IdempotencyKey, &locked.CommandKind,
		&locked.PartitionKey, &locked.SerializationKeyDigest, &locked.PayloadFormatVersion,
		&locked.RequestDigest, &locked.Payload, &locked.Receipts, &state, &token,
		&businessAttempt, &epoch, &expiresAt, &locked.PublicationIdentity,
		&publishAttempt, &encodedPending,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return rollback(ErrRuntimePlatformCommandLeaseLost)
		}
		return rollback(fmt.Errorf("runtime platform command publication: lock completion: %w", err))
	}
	locked.State = state
	locked.PublishAttempt = publishAttempt
	locked.ResultPayload, err = decodeNullableRuntimePlatformCommandResult(encodedPending)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: decode locked pending result: %w", err))
	}
	var now time.Time
	if err := tx.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: read completion time: %w", err))
	}
	if state != RuntimePlatformCommandPublishing || token != lease.LeaseToken ||
		businessAttempt <= 0 || publishAttempt != lease.Attempt ||
		epoch != s.expected.EpochDecimal() ||
		!runtimePlatformCommandDatabaseUTC(expiresAt).After(runtimePlatformCommandDatabaseUTC(now)) ||
		locked.PublicationIdentity != lease.Row.PublicationIdentity ||
		!bytes.Equal(locked.ResultPayload, lease.Row.ResultPayload) ||
		!runtimePlatformCommandSameImmutableAuthority(locked, lease.Row) {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'COMPLETED', ResultPayload = ?, Receipts = ?, LastError = NULL,
    RecoveryState = NULL, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND State = 'PUBLISHING' AND PublicationIdentity = ?
  AND LeaseToken = ? AND PublishAttempt = ? AND OwnershipEpoch = ?
  AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
		encodeRuntimePlatformCommandResult(terminalResult), receipts,
		lease.Row.RowID, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		lease.Row.PublicationIdentity, lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
	)
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: complete: %w", err))
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf(
			"%w: runtime platform command publication completion commit: %v",
			ErrRuntimePlatformCommandOutcomeUncertain,
			err,
		)
	}
	return nil
}

func (s *RuntimePlatformCommandExecutionStore) releasePublicationLease(
	ctx context.Context,
	lease RuntimePlatformCommandPublicationLease,
	message string,
	backoff time.Duration,
) error {
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("runtime platform command publication: begin release: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) error {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return errors.Join(cause, rollbackErr)
		}
		return cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	var result sql.Result
	if lease.Attempt >= s.maxPublicationAttempts {
		result, err = tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET State = 'POISONED', RecoveryState = 'PUBLISHING',
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PUBLISHING' AND PublicationIdentity = ?
  AND LeaseToken = ? AND PublishAttempt = ? AND OwnershipEpoch = ?
  AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
			message, lease.Row.RowID, lease.Row.PublicationIdentity,
			lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
		)
	} else {
		result, err = tx.ExecContext(ctx, `UPDATE _System_RuntimePlatformCommand
SET AvailableAt = TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)),
    LeaseToken = NULL, LeaseExpiresAt = NULL, OwnershipEpoch = NULL,
    LastError = ?, LastModifiedDate = UTC_TIMESTAMP(6)
WHERE Id = ? AND State = 'PUBLISHING' AND PublicationIdentity = ?
  AND LeaseToken = ? AND PublishAttempt = ? AND OwnershipEpoch = ?
  AND LeaseExpiresAt > UTC_TIMESTAMP(6)`,
			backoff.Microseconds(), message, lease.Row.RowID, lease.Row.PublicationIdentity,
			lease.LeaseToken, lease.Attempt, s.expected.EpochDecimal(),
		)
	}
	if err != nil {
		return rollback(fmt.Errorf("runtime platform command publication: release lease: %w", err))
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		return rollback(ErrRuntimePlatformCommandLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf(
			"%w: runtime platform command publication release commit: %v",
			ErrRuntimePlatformCommandOutcomeUncertain,
			err,
		)
	}
	return nil
}

func (s *RuntimePlatformCommandExecutionStore) validatePublicationLease(
	lease RuntimePlatformCommandPublicationLease,
) error {
	expectedIdentity, identityErr := runtimePlatformCommandPublicationIdentity(
		lease.Row.ScopeDigest, lease.Row.CommandID, lease.Row.RequestDigest,
	)
	if s == nil || s.db == nil || s.ownership == nil || identityErr != nil ||
		lease.Row.RowID == "" || lease.Row.State != RuntimePlatformCommandPublishing ||
		lease.Row.PublicationIdentity != expectedIdentity || len(lease.Row.ResultPayload) == 0 ||
		lease.LeaseToken == "" || lease.Attempt <= 0 ||
		lease.AttemptsRemaining != s.maxPublicationAttempts-lease.Attempt || lease.AttemptsRemaining < 0 ||
		lease.LeaseExpiresAt.IsZero() || lease.OwnershipEpoch != s.expected.OwnershipEpoch ||
		lease.Row.OrgID != s.scope.OrgID || lease.Row.AppInstallationID != s.scope.AppInstallationID ||
		lease.Row.RuntimeCapabilityID != s.scope.RuntimeCapabilityID ||
		lease.Row.ExecutionScopeDigest != s.scopeDigest {
		return errors.New("runtime platform command publication: lease is incomplete or outside configured authority")
	}
	return nil
}

func runtimePlatformCommandPublicationIdentity(scopeDigest, commandID, requestDigest string) (string, error) {
	if !hexDigestPattern.MatchString(scopeDigest) || commandID == "" || requestDigest == "" {
		return "", errors.New("runtime platform command publication: identity inputs are invalid")
	}
	hash := sha256.New()
	for _, value := range []string{scopeDigest, commandID, requestDigest} {
		var length [8]byte
		binary.BigEndian.PutUint64(length[:], uint64(len(value)))
		_, _ = hash.Write(length[:])
		_, _ = hash.Write([]byte(value))
	}
	return runtimePlatformCommandPublicationIdentityPrefix + hex.EncodeToString(hash.Sum(nil)), nil
}

func decodeNullableRuntimePlatformCommandResult(value sql.NullString) ([]byte, error) {
	if !value.Valid {
		return nil, nil
	}
	return decodeRuntimePlatformCommandResult(value.String)
}