package persistence

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

	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

const (
	RuntimeOutboundIntentPending   = "PENDING"
	RuntimeOutboundIntentAccepted  = "ACCEPTED"
	RuntimeOutboundIntentRejected  = "REJECTED"
	RuntimeOutboundIntentCompleted = "COMPLETED"
)

var (
	ErrRuntimeOutboundIntentConflict            = errors.New("runtime outbound intent identity is bound to different content")
	ErrRuntimeOutboundIntentTransactionRequired = errors.New("runtime outbound intent mutation requires an active caller-owned transaction")
	ErrRuntimeOutboundIntentNotFound            = errors.New("runtime outbound intent was not found")
)

var runtimeOutboundIntentRequiredColumns = map[string]string{
	"Id": "varchar", "OrgId": "varchar", "AppInstallationId": "varchar",
	"AuthorizedWorkloadId": "varchar", "ScopeDigest": "char", "CatalogRevision": "decimal(20,0)",
	"IntentId": "varchar", "IdempotencyKey": "varchar", "PayloadFormatVersion": "bigint",
	"PayloadDigest": "char", "Operation": "varchar", "SourceObject": "varchar",
	"SourceRecordId": "varchar", "ConnectorOwnerKey": "varchar", "FlowDeveloperName": "varchar",
	"FlowActionName": "varchar", "ActionSlotId": "varchar", "DestinationSlotId": "varchar",
	"Bindings": "json", "CanonicalPayload": "mediumblob", "AdmissionState": "varchar",
	"IntegrationCommandId": "varchar", "RejectionCode": "varchar", "ResultRemoteId": "varchar",
	"ResultFormatVersion": "bigint", "ResultDigest": "char", "CanonicalResult": "mediumblob",
	"CreatedDate": "datetime", "LastModifiedDate": "datetime",
}

var runtimeOutboundIntentGenerationRequiredColumns = map[string]string{
	"ScopeDigest": "char", "OrgId": "varchar", "AppInstallationId": "varchar",
	"AuthorizedWorkloadId": "varchar", "Generation": "decimal(20,0)",
	"CreatedDate": "datetime", "LastModifiedDate": "datetime",
}

type RuntimeOutboundIntentScope struct {
	OrgID                string
	AppInstallationID    string
	AuthorizedWorkloadID string
}

func (s RuntimeOutboundIntentScope) validate() error {
	for name, value := range map[string]string{
		"org ID": s.OrgID, "AppInstallation ID": s.AppInstallationID,
		"authorized workload ID": s.AuthorizedWorkloadID,
	} {
		if value == "" || strings.TrimSpace(value) != value {
			return fmt.Errorf("runtime outbound intent %s is invalid", name)
		}
	}
	return nil
}

func (s RuntimeOutboundIntentScope) digest() string {
	sum := sha256.Sum256([]byte(strings.Join([]string{s.OrgID, s.AppInstallationID, s.AuthorizedWorkloadID}, "\x00")))
	return hex.EncodeToString(sum[:])
}

type RuntimeOutboundIntentRow struct {
	RowID                string
	Scope                RuntimeOutboundIntentScope
	Intent               outboundintentv1.Intent
	AdmissionState       string
	IntegrationCommandID string
	RejectionCode        string
	ResultRemoteID       string
	ResultFormatVersion  uint32
	ResultDigest         string
	CanonicalResult      []byte
}

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

func NewRuntimeOutboundIntentRepository(db *sql.DB, transactions *TransactionManager) *RuntimeOutboundIntentRepository {
	return &RuntimeOutboundIntentRepository{db: db, transactions: transactions}
}

func VerifyRuntimeOutboundIntentSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return errors.New("runtime outbound intent schema: context and database are required")
	}
	if err := VerifyRequiredTableColumns(ctx, db, "_System_RuntimeOutboundIntent", runtimeOutboundIntentRequiredColumns); err != nil {
		return fmt.Errorf("runtime outbound intent schema: %w", err)
	}
	if err := VerifyRequiredTableColumns(ctx, db, "_System_RuntimeOutboundIntentGeneration", runtimeOutboundIntentGenerationRequiredColumns); err != nil {
		return fmt.Errorf("runtime outbound intent generation schema: %w", err)
	}
	if err := VerifyExactTableIndex(ctx, db, "_System_RuntimeOutboundIntentGeneration", "PRIMARY", true, []string{"ScopeDigest"}); err != nil {
		return fmt.Errorf("runtime outbound intent generation schema: %w", err)
	}
	for _, index := range []struct {
		name    string
		unique  bool
		columns []string
	}{
		{"ux_runtime_outbound_intent_scope", true, []string{"ScopeDigest", "IdempotencyKey"}},
		{"ux_runtime_outbound_intent_id", true, []string{"ScopeDigest", "IntentId"}},
		{"idx_runtime_outbound_intent_pending", false, []string{"ScopeDigest", "AdmissionState", "CatalogRevision", "Id"}},
		{runtimeOutboundIntentOwnerIndex, false, runtimeOutboundIntentOwnerIndexColumns},
	} {
		if err := VerifyExactTableIndex(ctx, db, "_System_RuntimeOutboundIntent", index.name, index.unique, index.columns); err != nil {
			return fmt.Errorf("runtime outbound intent schema: %w", err)
		}
	}
	return nil
}

// Enqueue writes an immutable provider-neutral intent through the transaction
// already owned by the invoking Runtime mutation/Flow. It never opens a second
// transaction, so a rolled-back business mutation cannot leave an orphaned
// external-effect request behind.
func (r *RuntimeOutboundIntentRepository) Enqueue(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
	intent outboundintentv1.Intent,
) (RuntimeOutboundIntentRow, bool, error) {
	if r == nil || r.db == nil || r.transactions == nil {
		return RuntimeOutboundIntentRow{}, false, errors.New("runtime outbound intent repository is unavailable")
	}
	if err := scope.validate(); err != nil {
		return RuntimeOutboundIntentRow{}, false, err
	}
	if err := outboundintentv1.ValidateIntent(intent); err != nil {
		return RuntimeOutboundIntentRow{}, false, err
	}
	exec, err := r.activeTransaction(ctx)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, err
	}
	if err := guardRuntimeOutboundEnqueue(ctx, exec, intent); err != nil {
		if !errors.Is(err, ErrRuntimeOutboundOwnerClosed) {
			return RuntimeOutboundIntentRow{}, false, err
		}
		// A deleted source cannot admit new work. An exact terminal intent is
		// immutable, however, and a late enqueue retry must remain idempotent.
		// NOWAIT preserves the record -> owner lock order even on this path.
		row, loadErr := loadRuntimeOutboundIntentByExactIdentity(ctx, exec, scope, "IntentId", intent.IntentID, " FOR UPDATE NOWAIT")
		if loadErr == nil && sameRuntimeOutboundIntent(row, scope, intent) &&
			(row.AdmissionState == RuntimeOutboundIntentCompleted || row.AdmissionState == RuntimeOutboundIntentRejected) {
			return row, true, nil
		}
		if loadErr != nil && !errors.Is(loadErr, ErrRuntimeOutboundIntentNotFound) {
			return RuntimeOutboundIntentRow{}, false, runtimeOutboundLifecycleReadError(loadErr)
		}
		return RuntimeOutboundIntentRow{}, false, err
	}
	bindings, err := json.Marshal(intent.Bindings)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, fmt.Errorf("encode runtime outbound intent bindings: %w", err)
	}
	result, err := exec.ExecContext(ctx, `INSERT IGNORE INTO _System_RuntimeOutboundIntent
(Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
 IntentId, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
 SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName, FlowActionName,
 ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload, AdmissionState)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'PENDING')`,
		intent.IntentID, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.digest(), intent.CatalogRevision, intent.IntentID, intent.IdempotencyKey,
		intent.PayloadFormatVersion, intent.PayloadDigest, string(intent.Operation),
		intent.SourceObject, intent.SourceRecordID, intent.ConnectorOwnerKey,
		intent.FlowDeveloperName, intent.FlowActionName, intent.ActionSlotID,
		intent.DestinationSlotID, bindings, intent.CanonicalPayload,
	)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, fmt.Errorf("enqueue runtime outbound intent: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, fmt.Errorf("enqueue runtime outbound intent rows affected: %w", err)
	}
	if affected == 1 {
		if err := advanceRuntimeOutboundIntentGeneration(ctx, exec, scope); err != nil {
			return RuntimeOutboundIntentRow{}, false, err
		}
	}
	row, err := r.load(ctx, exec, scope, intent.IntentID, intent.IdempotencyKey, false)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, err
	}
	if !sameRuntimeOutboundIntent(row, scope, intent) {
		return row, true, ErrRuntimeOutboundIntentConflict
	}
	return row, affected == 0, nil
}

// ReadGeneration performs one exact primary-key lookup. The generation is a
// scheduling cursor only; callers must still use ListPending and the durable
// acknowledgement path before any provider work can become eligible.
func (r *RuntimeOutboundIntentRepository) ReadGeneration(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
) (uint64, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil {
		return 0, errors.New("runtime outbound intent generation read is invalid")
	}
	var (
		orgID, appInstallationID, workloadID string
		generation                           uint64
	)
	err := r.db.QueryRowContext(ctx, `SELECT OrgId, AppInstallationId, AuthorizedWorkloadId, Generation
FROM _System_RuntimeOutboundIntentGeneration
WHERE ScopeDigest = ?`, scope.digest()).Scan(
		&orgID,
		&appInstallationID,
		&workloadID,
		&generation,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("read runtime outbound intent generation: %w", err)
	}
	if orgID != scope.OrgID || appInstallationID != scope.AppInstallationID ||
		workloadID != scope.AuthorizedWorkloadID {
		return 0, ErrRuntimeOutboundIntentConflict
	}
	return generation, nil
}

func advanceRuntimeOutboundIntentGeneration(
	ctx context.Context,
	exec *sql.Tx,
	scope RuntimeOutboundIntentScope,
) error {
	advance := func() (int64, error) {
		result, err := exec.ExecContext(ctx, `UPDATE _System_RuntimeOutboundIntentGeneration
SET Generation = Generation + 1, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE ScopeDigest = ? AND BINARY OrgId = BINARY ? AND BINARY AppInstallationId = BINARY ?
  AND BINARY AuthorizedWorkloadId = BINARY ?`,
			scope.digest(), scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		)
		if err != nil {
			return 0, fmt.Errorf("advance runtime outbound intent generation: %w", err)
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return 0, fmt.Errorf("advance runtime outbound intent generation rows affected: %w", err)
		}
		return affected, nil
	}

	affected, err := advance()
	if err != nil {
		return err
	}
	if affected == 1 {
		return nil
	}
	if affected != 0 {
		return ErrRuntimeOutboundIntentConflict
	}

	result, err := exec.ExecContext(ctx, `INSERT IGNORE INTO _System_RuntimeOutboundIntentGeneration
(ScopeDigest, OrgId, AppInstallationId, AuthorizedWorkloadId, Generation)
VALUES (?, ?, ?, ?, 1)`, scope.digest(), scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID)
	if err != nil {
		return fmt.Errorf("initialize runtime outbound intent generation: %w", err)
	}
	inserted, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("initialize runtime outbound intent generation rows affected: %w", err)
	}
	if inserted == 1 {
		return nil
	}
	if inserted != 0 {
		return ErrRuntimeOutboundIntentConflict
	}

	// A concurrent first writer may have initialized the same trusted scope.
	// Re-run the exact-scope update; a digest collision still fails closed.
	affected, err = advance()
	if err != nil {
		return err
	}
	if affected != 1 {
		return ErrRuntimeOutboundIntentConflict
	}
	return nil
}

// ListPending returns only intents from the exact request-bound Catalog
// revision. It does not reinterpret older rows through current metadata.
func (r *RuntimeOutboundIntentRepository) ListPending(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
	catalogRevision uint64,
	limit uint32,
) ([]RuntimeOutboundIntentRow, error) {
	return r.ListPendingForOperationSet(ctx, scope, catalogRevision, limit, outboundintentv1.CurrentOperationSetVersion)
}

func (r *RuntimeOutboundIntentRepository) ListPendingForOperationSet(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
	catalogRevision uint64,
	limit uint32,
	maxOperationSetVersion uint32,
) ([]RuntimeOutboundIntentRow, error) {
	if r == nil || r.db == nil || scope.validate() != nil || catalogRevision == 0 || limit == 0 ||
		limit > outboundintentv1.MaxIntentsPerList || maxOperationSetVersion == 0 ||
		maxOperationSetVersion > outboundintentv1.CurrentOperationSetVersion {
		return nil, errors.New("runtime outbound intent list is invalid")
	}
	operationFence := ""
	args := []any{scope.digest(), scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID, catalogRevision}
	switch maxOperationSetVersion {
	case outboundintentv1.OperationSetVersionV1:
		operationFence = " AND Operation <> 'SEND_MAIL'"
	case outboundintentv1.OperationSetVersionV2:
		operationFence = " AND NOT (Operation = 'SEND_MAIL' AND PayloadFormatVersion = ?)"
		args = append(args, outboundintentv1.MailPayloadFormatV3)
	}
	args = append(args, limit)
	rows, err := r.db.QueryContext(ctx, runtimeOutboundIntentSelect+`
 WHERE ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
   AND AdmissionState = 'PENDING' AND CatalogRevision = ?`+operationFence+`
 ORDER BY Id LIMIT ?`, args...)
	if err != nil {
		return nil, fmt.Errorf("list pending runtime outbound intents: %w", err)
	}
	defer rows.Close()
	result := make([]RuntimeOutboundIntentRow, 0, limit)
	for rows.Next() {
		row, scanErr := scanRuntimeOutboundIntent(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		if row.Scope != scope || row.Intent.CatalogRevision != catalogRevision || row.AdmissionState != RuntimeOutboundIntentPending {
			return nil, ErrRuntimeOutboundIntentConflict
		}
		result = append(result, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list pending runtime outbound intent rows: %w", err)
	}
	return result, nil
}

func (r *RuntimeOutboundIntentRepository) Acknowledge(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
	request outboundintentv1.AcknowledgeRequest,
) (RuntimeOutboundIntentRow, bool, error) {
	if r == nil || r.db == nil || r.transactions == nil || scope.validate() != nil {
		return RuntimeOutboundIntentRow{}, false, errors.New("runtime outbound intent acknowledgment is unavailable")
	}
	exec, err := r.activeTransaction(ctx)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, err
	}
	row, err := r.load(ctx, exec, scope, request.IntentID, "", true)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, err
	}
	if outboundintentv1.ValidateAcknowledgeRequestForIntent(row.Intent, request) != nil {
		return row, false, ErrRuntimeOutboundIntentConflict
	}
	wantState := string(request.Outcome)
	if row.AdmissionState != RuntimeOutboundIntentPending {
		// A lost acknowledgement response may be retried after Integration has
		// already committed the provider result. COMPLETED preserves the same
		// accepted command identity, so the acknowledgement remains an
		// idempotent duplicate rather than reopening admission.
		if row.AdmissionState == RuntimeOutboundIntentCompleted &&
			request.Outcome == outboundintentv1.AdmissionAccepted &&
			row.IntegrationCommandID == request.IntegrationCommandID {
			return row, true, nil
		}
		if row.AdmissionState == wantState && row.IntegrationCommandID == request.IntegrationCommandID && row.RejectionCode == request.RejectionCode {
			return row, true, nil
		}
		return row, true, ErrRuntimeOutboundIntentConflict
	}
	result, err := exec.ExecContext(ctx, `UPDATE _System_RuntimeOutboundIntent
 SET AdmissionState = ?, IntegrationCommandId = NULLIF(?, ''), RejectionCode = NULLIF(?, ''),
     LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE Id = ? AND ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
   AND CatalogRevision = ? AND PayloadDigest = ? AND AdmissionState = 'PENDING'`,
		wantState, request.IntegrationCommandID, request.RejectionCode, row.RowID, scope.digest(),
		scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		request.CatalogRevision, request.PayloadDigest,
	)
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, fmt.Errorf("acknowledge runtime outbound intent: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return RuntimeOutboundIntentRow{}, false, fmt.Errorf("acknowledge runtime outbound intent rows affected: %w", err)
	}
	if affected != 1 {
		return RuntimeOutboundIntentRow{}, false, ErrRuntimeOutboundIntentConflict
	}
	row.AdmissionState = wantState
	row.IntegrationCommandID = request.IntegrationCommandID
	row.RejectionCode = request.RejectionCode
	return row, false, nil
}

// OpenMailAttachmentReference resolves one attachment only from the exact
// accepted intent/command fence. File identity is never caller-selected.
func (r *RuntimeOutboundIntentRepository) OpenMailAttachmentReference(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
	request outboundintentv1.OpenMailAttachmentRequest,
) (outboundintentv1.MailAttachmentV1, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil {
		return outboundintentv1.MailAttachmentV1{}, errors.New("runtime outbound mail attachment read is unavailable")
	}
	if err := outboundintentv1.ValidateOpenMailAttachmentRequest(ctx, request); err != nil {
		return outboundintentv1.MailAttachmentV1{}, err
	}
	row, err := r.load(ctx, r.db, scope, request.IntentID, "", false)
	if err != nil {
		return outboundintentv1.MailAttachmentV1{}, err
	}
	if row.AdmissionState != RuntimeOutboundIntentAccepted {
		return outboundintentv1.MailAttachmentV1{}, ErrRuntimeOutboundIntentConflict
	}
	attachment, err := outboundintentv1.ValidateOpenMailAttachmentRequestForAcceptedIntent(
		row.Intent,
		row.IntegrationCommandID,
		request,
	)
	if err != nil {
		return outboundintentv1.MailAttachmentV1{}, ErrRuntimeOutboundIntentConflict
	}
	return attachment, nil
}

// RuntimeOutboundResultProjection applies the Runtime-local consequence of a
// provider result. CommitResultWithProjection invokes it only after locking and
// verifying the exact ACCEPTED Integration command, and before it records the
// terminal receipt in the same caller-owned transaction.
type RuntimeOutboundResultProjection func(context.Context, RuntimeOutboundIntentRow) error

// CommitResultWithProjection closes accepted-command authorization, local
// projection and the terminal receipt as one operation. The completion update
// is private to this method, so callers cannot mark an intent COMPLETED without
// first returning successfully from the Runtime-owned projection callback.
// Provider I/O must have completed before entering this method.
func (r *RuntimeOutboundIntentRepository) CommitResultWithProjection(
	ctx context.Context,
	scope RuntimeOutboundIntentScope,
	request outboundintentv1.CommitResultRequest,
	projection RuntimeOutboundResultProjection,
) (bool, error) {
	if r == nil || r.db == nil || r.transactions == nil || scope.validate() != nil || projection == nil {
		return false, errors.New("runtime outbound intent result commit is unavailable")
	}
	exec, err := r.activeTransaction(ctx)
	if err != nil {
		return false, err
	}
	row, err := r.load(ctx, exec, scope, request.IntentID, "", true)
	if err != nil {
		return false, err
	}
	if outboundintentv1.ValidateCommitResultRequestForIntent(row.Intent, request) != nil ||
		row.IntegrationCommandID != request.IntegrationCommandID {
		return false, ErrRuntimeOutboundIntentConflict
	}
	switch row.AdmissionState {
	case RuntimeOutboundIntentAccepted:
		if row.ResultRemoteID != "" || row.ResultFormatVersion != 0 ||
			row.ResultDigest != "" || len(row.CanonicalResult) != 0 {
			return false, ErrRuntimeOutboundIntentConflict
		}
	case RuntimeOutboundIntentCompleted:
		if sameRuntimeOutboundResult(row, request) {
			return true, nil
		}
		return false, ErrRuntimeOutboundIntentConflict
	default:
		return false, ErrRuntimeOutboundIntentConflict
	}
	if err := projection(ctx, row); err != nil {
		return false, err
	}
	result, err := exec.ExecContext(ctx, `UPDATE _System_RuntimeOutboundIntent
 SET AdmissionState = 'COMPLETED', ResultRemoteId = NULLIF(?, ''),
     ResultFormatVersion = NULLIF(?, 0), ResultDigest = NULLIF(?, ''),
     CanonicalResult = NULLIF(?, ''), LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE Id = ? AND ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
   AND CatalogRevision = ? AND PayloadDigest = ? AND IntegrationCommandId = ?
   AND AdmissionState = 'ACCEPTED' AND ResultRemoteId IS NULL
   AND ResultFormatVersion IS NULL AND ResultDigest IS NULL AND CanonicalResult IS NULL`,
		request.RemoteID, request.ResultFormatVersion, request.ResultDigest, request.CanonicalResult,
		request.IntentID, scope.digest(), scope.OrgID, scope.AppInstallationID,
		scope.AuthorizedWorkloadID, request.CatalogRevision, request.PayloadDigest,
		request.IntegrationCommandID,
	)
	if err != nil {
		return false, fmt.Errorf("complete runtime outbound intent result: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return false, fmt.Errorf("complete runtime outbound intent result rows affected: %w", err)
	}
	if affected != 1 {
		return false, ErrRuntimeOutboundIntentConflict
	}
	return false, nil
}

const runtimeOutboundIntentSelect = `SELECT
 Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
 IntentId, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
 SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName, FlowActionName,
 ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload, AdmissionState,
 COALESCE(IntegrationCommandId, ''), COALESCE(RejectionCode, ''), COALESCE(ResultRemoteId, ''),
 COALESCE(ResultFormatVersion, 0), COALESCE(ResultDigest, ''), COALESCE(CanonicalResult, '')
 FROM _System_RuntimeOutboundIntent`

type runtimeOutboundIntentScanner interface{ Scan(...any) error }

func scanRuntimeOutboundIntent(scanner runtimeOutboundIntentScanner) (RuntimeOutboundIntentRow, error) {
	var row RuntimeOutboundIntentRow
	var bindings []byte
	var operation string
	var scopeDigest string
	if err := scanner.Scan(
		&row.RowID, &row.Scope.OrgID, &row.Scope.AppInstallationID, &row.Scope.AuthorizedWorkloadID, &scopeDigest,
		&row.Intent.CatalogRevision, &row.Intent.IntentID, &row.Intent.IdempotencyKey,
		&row.Intent.PayloadFormatVersion, &row.Intent.PayloadDigest, &operation,
		&row.Intent.SourceObject, &row.Intent.SourceRecordID, &row.Intent.ConnectorOwnerKey,
		&row.Intent.FlowDeveloperName, &row.Intent.FlowActionName, &row.Intent.ActionSlotID,
		&row.Intent.DestinationSlotID, &bindings, &row.Intent.CanonicalPayload,
		&row.AdmissionState, &row.IntegrationCommandID, &row.RejectionCode, &row.ResultRemoteID,
		&row.ResultFormatVersion, &row.ResultDigest, &row.CanonicalResult,
	); err != nil {
		return RuntimeOutboundIntentRow{}, fmt.Errorf("scan runtime outbound intent: %w", err)
	}
	if scopeDigest != row.Scope.digest() {
		return RuntimeOutboundIntentRow{}, ErrRuntimeOutboundIntentConflict
	}
	row.Intent.Operation = outboundintentv1.Operation(operation)
	if err := json.Unmarshal(bindings, &row.Intent.Bindings); err != nil || outboundintentv1.ValidateIntent(row.Intent) != nil {
		return RuntimeOutboundIntentRow{}, ErrRuntimeOutboundIntentConflict
	}
	return row, nil
}

func (r *RuntimeOutboundIntentRepository) load(
	ctx context.Context,
	exec Executor,
	scope RuntimeOutboundIntentScope,
	intentID string,
	idempotencyKey string,
	forUpdate bool,
) (RuntimeOutboundIntentRow, error) {
	lockClause := ""
	if forUpdate {
		lockClause = " FOR UPDATE"
	}
	row, err := loadRuntimeOutboundIntentByExactIdentity(
		ctx, exec, scope, "IntentId", intentID, lockClause,
	)
	if err == nil || idempotencyKey == "" || !errors.Is(err, ErrRuntimeOutboundIntentNotFound) {
		return row, err
	}
	return loadRuntimeOutboundIntentByExactIdentity(
		ctx, exec, scope, "IdempotencyKey", idempotencyKey, lockClause,
	)
}

func loadRuntimeOutboundIntentByExactIdentity(
	ctx context.Context,
	exec Executor,
	scope RuntimeOutboundIntentScope,
	column string,
	value string,
	lockClause string,
) (RuntimeOutboundIntentRow, error) {
	if (column != "IntentId" && column != "IdempotencyKey") || value == "" {
		return RuntimeOutboundIntentRow{}, ErrRuntimeOutboundIntentNotFound
	}
	query := runtimeOutboundIntentSelect + `
 WHERE ScopeDigest = ? AND ` + column + ` = ?
   AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
	LIMIT 2`
	switch lockClause {
	case "", " FOR UPDATE", " FOR UPDATE NOWAIT":
		query += lockClause
	default:
		return RuntimeOutboundIntentRow{}, ErrRuntimeOutboundIntentConflict
	}
	rows, err := exec.QueryContext(
		ctx, query, scope.digest(), value,
		scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
	)
	if err != nil {
		return RuntimeOutboundIntentRow{}, fmt.Errorf("load runtime outbound intent by %s: %w", column, err)
	}
	defer rows.Close()
	var result RuntimeOutboundIntentRow
	count := 0
	for rows.Next() {
		row, scanErr := scanRuntimeOutboundIntent(rows)
		if scanErr != nil {
			return RuntimeOutboundIntentRow{}, scanErr
		}
		result = row
		count++
	}
	if err := rows.Err(); err != nil {
		return RuntimeOutboundIntentRow{}, fmt.Errorf("load runtime outbound intent by %s rows: %w", column, err)
	}
	if count == 0 {
		return RuntimeOutboundIntentRow{}, ErrRuntimeOutboundIntentNotFound
	}
	if count != 1 {
		return RuntimeOutboundIntentRow{}, ErrRuntimeOutboundIntentConflict
	}
	return result, nil
}

func (r *RuntimeOutboundIntentRepository) activeTransaction(ctx context.Context) (*sql.Tx, error) {
	if r == nil || r.transactions == nil {
		return nil, ErrRuntimeOutboundIntentTransactionRequired
	}
	tx, ok := r.transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return nil, ErrRuntimeOutboundIntentTransactionRequired
	}
	return tx, nil
}

func sameRuntimeOutboundIntent(row RuntimeOutboundIntentRow, scope RuntimeOutboundIntentScope, intent outboundintentv1.Intent) bool {
	if row.Scope != scope || row.RowID != intent.IntentID || row.Intent.IntentID != intent.IntentID ||
		row.Intent.IdempotencyKey != intent.IdempotencyKey || row.Intent.PayloadFormatVersion != intent.PayloadFormatVersion ||
		row.Intent.PayloadDigest != intent.PayloadDigest || row.Intent.CatalogRevision != intent.CatalogRevision ||
		row.Intent.Operation != intent.Operation || row.Intent.SourceObject != intent.SourceObject ||
		row.Intent.SourceRecordID != intent.SourceRecordID || row.Intent.ConnectorOwnerKey != intent.ConnectorOwnerKey ||
		row.Intent.FlowDeveloperName != intent.FlowDeveloperName || row.Intent.FlowActionName != intent.FlowActionName ||
		row.Intent.ActionSlotID != intent.ActionSlotID || row.Intent.DestinationSlotID != intent.DestinationSlotID ||
		string(row.Intent.CanonicalPayload) != string(intent.CanonicalPayload) {
		return false
	}
	left, leftErr := json.Marshal(row.Intent.Bindings)
	right, rightErr := json.Marshal(intent.Bindings)
	return leftErr == nil && rightErr == nil && string(left) == string(right)
}

func sameRuntimeOutboundResult(row RuntimeOutboundIntentRow, request outboundintentv1.CommitResultRequest) bool {
	return row.ResultRemoteID == request.RemoteID &&
		row.ResultFormatVersion == request.ResultFormatVersion &&
		row.ResultDigest == request.ResultDigest &&
		bytes.Equal(row.CanonicalResult, request.CanonicalResult)
}