package connector_outboundintent

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

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type Repository struct {
	db                        *sql.DB
	now                       func() time.Time
	typedResultSchemaVerified atomic.Bool
}

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

var commandRequiredColumns = map[string]string{
	"Id": "varchar", "AcceptanceSequence": "bigint", "OrgId": "varchar", "AppInstallationId": "varchar",
	"AuthorizedWorkloadId": "varchar", "ScopeDigest": "char", "CatalogRevision": "decimal",
	"IntentId": "varchar", "LinkKey": "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", "State": "varchar",
	"AttemptCount": "bigint", "RetryCount": "bigint", "ClaimedBy": "varchar", "ClaimToken": "varchar",
	"ClaimedAt": "datetime", "LeaseExpiresAt": "datetime", "DispatchStartedAt": "datetime",
	"NextAttemptAt": "datetime", "DependencyKind": "varchar", "DependencyKey": "varchar",
	"DependencyReadinessKeys": "json", "DependencyBlockedAt": "datetime", "FailureCode": "varchar", "ResponseStatus": "bigint",
	"RemoteId": "varchar", "CreatedDate": "datetime", "LastModifiedDate": "datetime",
}

var typedResultColumns = map[string]string{
	"ResultFormatVersion": "bigint", "ResultDigest": "char", "CanonicalResult": "mediumblob",
}

type storedTypedTerminalResult struct {
	formatVersion uint32
	digest        string
	canonical     []byte
}

var remoteLinkRequiredColumns = map[string]string{
	"Id": "varchar", "LinkKey": "varchar", "Status": "varchar", "RemoteId": "varchar",
	"RemoteLocatorHash": "varchar", "RemoteLocatorActiveKey": "varchar",
	"ConnectorOwnerKey": "varchar", "ActionSlotId": "varchar", "ActionType": "varchar",
	"SourceObject": "varchar", "SourceRecordId": "varchar", "DestinationSlotId": "varchar",
	"RequesterOwnerPack": "varchar", "ProfileOwnerPack": "varchar", "ConnectorNamespace": "varchar",
	"FlowDeveloperName": "varchar", "FlowActionName": "varchar", "TargetObject": "varchar",
	"ProfileName": "varchar", "CreatedPackInstallId": "varchar", "LastValidatedPackInstallId": "varchar",
	"CredentialRef": "varchar", "EndpointClass": "varchar", "RetryPolicy": "longtext",
	"ProfileContractHash": "varchar", "CreatePathTemplate": "varchar", "UpdatePathTemplate": "varchar",
	"ResponseIdPath": "varchar", "RemoteIdField": "varchar", "RemoteIdProvenanceField": "varchar",
	"MappingFingerprint": "varchar", "BodyFingerprint": "varchar", "CompatibilityMirrorStatus": "varchar",
	"CreatedDate": "datetime", "LastModifiedDate": "datetime",
}

func (r *Repository) VerifySchema(ctx context.Context) error {
	if r == nil || r.db == nil || ctx == nil {
		return ErrDependency
	}
	r.typedResultSchemaVerified.Store(false)
	if err := verifyColumns(ctx, r.db, "_System_IntegrationOutboundCommand", commandRequiredColumns); err != nil {
		return err
	}
	if err := verifyColumnCollation(
		ctx, r.db, "_System_IntegrationOutboundCommand", "DependencyKey", "utf8mb4_bin",
	); err != nil {
		return err
	}
	if err := verifyColumnExtra(ctx, r.db, "_System_IntegrationOutboundCommand", "AcceptanceSequence", "auto_increment"); err != nil {
		return err
	}
	if err := verifyColumns(ctx, r.db, "_System_ConnectorRemoteLink", remoteLinkRequiredColumns); err != nil {
		return err
	}
	for _, index := range []struct {
		table   string
		name    string
		unique  bool
		columns []string
	}{
		{"_System_IntegrationOutboundCommand", "ux_integration_outbound_intent", true, []string{"ScopeDigest", "IntentId"}},
		{"_System_IntegrationOutboundCommand", "ux_integration_outbound_idempotency", true, []string{"ScopeDigest", "IdempotencyKey"}},
		{"_System_IntegrationOutboundCommand", "ux_integration_outbound_acceptance_sequence", true, []string{"AcceptanceSequence"}},
		{"_System_IntegrationOutboundCommand", "idx_integration_outbound_claim", false, []string{"ScopeDigest", "CatalogRevision", "State", "NextAttemptAt", "Id"}},
		{"_System_IntegrationOutboundCommand", "idx_integration_outbound_lane", false, []string{"ScopeDigest", "CatalogRevision", "LinkKey", "State", "AcceptanceSequence"}},
		{"_System_IntegrationOutboundCommand", "idx_integration_outbound_dependency", false, []string{"ScopeDigest", "DependencyKind", "DependencyKey", "State", "Id"}},
		{"_System_IntegrationOutboundCommand", "idx_integration_outbound_dependency_reconcile", false, []string{"ScopeDigest", "CatalogRevision", "State", "Id"}},
		{"_System_ConnectorRemoteLink", "uniq_connector_remote_link_key", true, []string{"LinkKey"}},
		{"_System_ConnectorRemoteLink", "uniq_connector_remote_locator_active", true, []string{"RemoteLocatorActiveKey"}},
	} {
		if err := verifyIndex(ctx, r.db, index.table, index.name, index.unique, index.columns); err != nil {
			return err
		}
	}
	typedResults, err := verifyOptionalColumns(ctx, r.db, "_System_IntegrationOutboundCommand", typedResultColumns)
	if err != nil {
		return err
	}
	r.typedResultSchemaVerified.Store(typedResults)
	return nil
}

func verifyColumnCollation(ctx context.Context, db *sql.DB, table, column, required string) error {
	var collation string
	err := db.QueryRowContext(ctx, `
SELECT COLLATION_NAME
  FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?
 LIMIT 1`, table, column).Scan(&collation)
	if err != nil {
		return fmt.Errorf("connector outbound intent verify %s.%s collation: %w", table, column, err)
	}
	if !strings.EqualFold(collation, required) {
		return fmt.Errorf(
			"connector outbound intent schema %s column %s collation is %q, want %q",
			table, column, collation, required,
		)
	}
	return nil
}

// typedResultsAvailable lazily detects the all-or-nothing schema extension so
// a Repository reconstructed after process restart can resume typed writeback
// without relying on an earlier in-memory VerifySchema call. Legacy schemas
// remain ready for legacy effects.
func (r *Repository) typedResultsAvailable(ctx context.Context) (bool, error) {
	if r == nil || r.db == nil || ctx == nil {
		return false, ErrDependency
	}
	if r.typedResultSchemaVerified.Load() {
		return true, nil
	}
	available, err := verifyOptionalColumns(ctx, r.db, "_System_IntegrationOutboundCommand", typedResultColumns)
	if err != nil {
		return false, err
	}
	if available {
		r.typedResultSchemaVerified.Store(true)
	}
	return available, nil
}

func verifyColumnExtra(ctx context.Context, db *sql.DB, table, column, required string) error {
	var extra string
	err := db.QueryRowContext(ctx, `
SELECT EXTRA
  FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?
 LIMIT 1`, table, column).Scan(&extra)
	if err != nil {
		return fmt.Errorf("connector outbound intent verify %s.%s column semantics: %w", table, column, err)
	}
	for _, semantic := range strings.Fields(strings.ToLower(extra)) {
		if semantic == strings.ToLower(required) {
			return nil
		}
	}
	return fmt.Errorf("connector outbound intent schema %s column %s extra is %q, want %q", table, column, extra, required)
}

func verifyColumns(ctx context.Context, db *sql.DB, table string, required map[string]string) error {
	found, err := loadColumnTypes(ctx, db, table)
	if err != nil {
		return err
	}
	for name, dataType := range required {
		if found[name] != dataType {
			return fmt.Errorf("connector outbound intent schema %s column %s is %q, want %q", table, name, found[name], dataType)
		}
	}
	return nil
}

func verifyOptionalColumns(ctx context.Context, db *sql.DB, table string, optional map[string]string) (bool, error) {
	found, err := loadColumnTypes(ctx, db, table)
	if err != nil {
		return false, err
	}
	return classifyOptionalColumns(table, found, optional)
}

func classifyOptionalColumns(table string, found, optional map[string]string) (bool, error) {
	present := 0
	for name, dataType := range optional {
		if found[name] == "" {
			continue
		}
		present++
		if found[name] != dataType {
			return false, fmt.Errorf("connector outbound intent schema %s optional column %s is %q, want %q", table, name, found[name], dataType)
		}
	}
	if present != 0 && present != len(optional) {
		return false, fmt.Errorf("connector outbound intent schema %s has a partial typed-result extension", table)
	}
	return present == len(optional), nil
}

func loadColumnTypes(ctx context.Context, db *sql.DB, table string) (map[string]string, error) {
	rows, err := db.QueryContext(ctx, `
SELECT COLUMN_NAME, DATA_TYPE
  FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?`, table)
	if err != nil {
		return nil, fmt.Errorf("connector outbound intent verify %s columns: %w", table, err)
	}
	defer rows.Close()
	found := map[string]string{}
	for rows.Next() {
		var name, dataType string
		if err := rows.Scan(&name, &dataType); err != nil {
			return nil, err
		}
		found[name] = strings.ToLower(dataType)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return found, nil
}

func verifyIndex(ctx context.Context, db *sql.DB, table, name string, unique bool, expected []string) error {
	rows, err := db.QueryContext(ctx, `
SELECT COLUMN_NAME, NON_UNIQUE
  FROM INFORMATION_SCHEMA.STATISTICS
 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
   AND INDEX_NAME = ?
 ORDER BY SEQ_IN_INDEX`, table, name)
	if err != nil {
		return fmt.Errorf("connector outbound intent verify index %s.%s: %w", table, name, err)
	}
	defer rows.Close()
	var columns []string
	nonUnique := -1
	for rows.Next() {
		var column string
		var current int
		if err := rows.Scan(&column, &current); err != nil {
			return err
		}
		if nonUnique == -1 {
			nonUnique = current
		} else if nonUnique != current {
			return fmt.Errorf("connector outbound intent index %s.%s uniqueness is mixed", table, name)
		}
		columns = append(columns, column)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if !reflect.DeepEqual(columns, expected) || (nonUnique == 0) != unique {
		return fmt.Errorf("connector outbound intent index %s.%s is columns %v unique=%v, want %v unique=%v", table, name, columns, nonUnique == 0, expected, unique)
	}
	return nil
}

func commandID(scope Scope, intent outboundintentv1.Intent) string {
	sum := sha256.Sum256([]byte(strings.Join([]string{
		scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		fmt.Sprintf("%d", scope.CatalogRevision), intent.IntentID,
	}, "\x00")))
	return "integration-outbound-" + hex.EncodeToString(sum[:])
}

func integrationLogicalEffectScope(scope Scope) outboundintentv1.LogicalEffectScope {
	return outboundintentv1.LogicalEffectScope{
		OrgID: scope.OrgID, AppInstallationID: scope.AppInstallationID,
		AuthorizedWorkloadID: scope.AuthorizedWorkloadID,
	}
}

func scopeDigest(scope Scope) string {
	return outboundintentv1.IntegrationScopeDigest(integrationLogicalEffectScope(scope))
}

func outboundLinkKey(scope Scope, intent outboundintentv1.Intent) string {
	return linkKeyForIdentity(
		scope,
		intent.ConnectorOwnerKey, intent.ActionSlotID, intent.SourceObject,
		intent.SourceRecordID, intent.DestinationSlotID,
	)
}

func linkKeyForIdentity(scope Scope, connectorOwnerKey, actionSlotID, sourceObject, sourceRecordID, destinationSlotID string) string {
	return outboundintentv1.IntegrationLinkKeyV2(integrationLogicalEffectScope(scope), connectorOwnerKey,
		actionSlotID, sourceObject, sourceRecordID, destinationSlotID)
}

func (r *Repository) Accept(ctx context.Context, scope Scope, intent outboundintentv1.Intent) (Command, bool, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil ||
		outboundintentv1.ValidateIntent(intent) != nil || intent.CatalogRevision != scope.CatalogRevision {
		return Command{}, false, ErrInvalidInput
	}
	bindings, err := json.Marshal(intent.Bindings)
	if err != nil {
		return Command{}, false, ErrInvalidInput
	}
	now := r.now().UTC()
	id := commandID(scope, intent)
	if intent.Operation != outboundintentv1.OperationUpsert {
		return r.acceptWithoutLegacyFence(ctx, scope, intent, bindings, now, id)
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, false, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, bool, error) {
		return Command{}, false, errors.Join(cause, tx.Rollback())
	}
	result, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO _System_IntegrationOutboundCommand
       (Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
        IntentId, LinkKey, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
        SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName,
        FlowActionName, ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload,
        State, AttemptCount, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
		id, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scopeDigest(scope), scope.CatalogRevision, intent.IntentID, outboundLinkKey(scope, intent), 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, StatePending, now, now)
	if err != nil {
		return rollback(fmt.Errorf("connector outbound intent accept: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return rollback(err)
	}
	command, err := loadByIdentityForUpdateFrom(ctx, tx, scope, intent.IntentID, intent.IdempotencyKey)
	if err != nil {
		return rollback(err)
	}
	if !sameCommand(command, scope, intent) {
		duplicate := affected == 0
		rollbackErr := tx.Rollback()
		return Command{}, duplicate, errors.Join(ErrIdentityConflict, rollbackErr)
	}
	duplicate := affected == 0
	if !duplicate {
		_, legacyFound, lockErr := lockRemoteLink(ctx, tx, legacyRemoteLinkKey(command))
		if lockErr != nil && !isMissingRemoteLinkTable(lockErr) {
			return rollback(lockErr)
		}
		if lockErr == nil && legacyFound {
			if err := rejectForeignLegacyDispatches(ctx, tx, command); err != nil {
				return rollback(err)
			}
		}
	}
	if err := tx.Commit(); err != nil {
		return Command{}, false, err
	}
	return command, duplicate, nil
}

func (r *Repository) acceptWithoutLegacyFence(
	ctx context.Context,
	scope Scope,
	intent outboundintentv1.Intent,
	bindings []byte,
	now time.Time,
	id string,
) (Command, bool, error) {
	result, err := r.db.ExecContext(ctx, `
INSERT IGNORE INTO _System_IntegrationOutboundCommand
       (Id, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
        IntentId, LinkKey, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
        SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName,
        FlowActionName, ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload,
        State, AttemptCount, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
		id, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scopeDigest(scope), scope.CatalogRevision, intent.IntentID, outboundLinkKey(scope, intent), 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, StatePending, now, now)
	if err != nil {
		return Command{}, false, fmt.Errorf("connector outbound intent accept: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return Command{}, false, err
	}
	command, err := r.loadByIdentity(ctx, scope, intent.IntentID, intent.IdempotencyKey)
	if err != nil {
		return Command{}, false, err
	}
	if !sameCommand(command, scope, intent) {
		return Command{}, affected == 0, ErrIdentityConflict
	}
	return command, affected == 0, nil
}

func isMissingRemoteLinkTable(err error) bool {
	var mysqlErr *mysql.MySQLError
	return errors.As(err, &mysqlErr) && mysqlErr.Number == 1146 &&
		strings.Contains(strings.ToLower(mysqlErr.Message), "_system_connectorremotelink")
}

const outboundWorkHintSQL = `
SELECT
  EXISTS (
    SELECT 1
      FROM _System_IntegrationOutboundCommand AS writeback
     WHERE writeback.ScopeDigest = ?
       AND writeback.OrgId = ?
       AND writeback.AppInstallationId = ?
       AND writeback.AuthorizedWorkloadId = ?
       AND writeback.CatalogRevision = ?
       AND writeback.State = ?
       AND (writeback.NextAttemptAt IS NULL OR writeback.NextAttemptAt <= ?)
       AND (writeback.ClaimToken IS NULL OR writeback.ClaimToken = '' OR writeback.LeaseExpiresAt <= ?)
     LIMIT 1
  ) OR EXISTS (
    SELECT 1
      FROM _System_IntegrationOutboundCommand AS expired
     WHERE expired.ScopeDigest = ?
       AND expired.OrgId = ?
       AND expired.AppInstallationId = ?
       AND expired.AuthorizedWorkloadId = ?
       AND expired.CatalogRevision = ?
       AND expired.State IN (?, ?)
       AND expired.LeaseExpiresAt <= ?
     LIMIT 1
  ) OR EXISTS (
    SELECT 1
      FROM _System_IntegrationOutboundCommand AS candidate
     WHERE candidate.ScopeDigest = ?
       AND candidate.OrgId = ?
       AND candidate.AppInstallationId = ?
       AND candidate.AuthorizedWorkloadId = ?
       AND candidate.CatalogRevision = ?
       AND candidate.State IN (?, ?, ?)
       AND (candidate.NextAttemptAt IS NULL OR candidate.NextAttemptAt <= ?)
       AND NOT EXISTS (
         SELECT 1
           FROM _System_IntegrationOutboundCommand AS prior
          WHERE prior.ScopeDigest = candidate.ScopeDigest
            AND prior.OrgId = candidate.OrgId
            AND prior.AppInstallationId = candidate.AppInstallationId
            AND prior.AuthorizedWorkloadId = candidate.AuthorizedWorkloadId
            AND prior.CatalogRevision = candidate.CatalogRevision
            AND prior.LinkKey = candidate.LinkKey
            AND prior.State IN (?, ?, ?, ?, ?, ?, ?, ?)
            AND prior.AcceptanceSequence < candidate.AcceptanceSequence
       )
     LIMIT 1
  )`

// HasWorkHint is a negative-only, read-only admission hint for the durable
// outbound worker. A positive result never authorizes an effect: ClaimWriteback
// and ClaimNext retain their complete pessimistic transactions, ordering and
// lease recovery. A racing insert after a negative result is visible on the
// next bounded worker poll.
func (r *Repository) HasWorkHint(ctx context.Context, scope Scope) (bool, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil {
		return false, ErrInvalidInput
	}
	now := r.now().UTC()
	digest := scopeDigest(scope)
	var found bool
	err := r.db.QueryRowContext(
		ctx,
		outboundWorkHintSQL,
		digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StateWriteback, now, now,
		digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StateClaimed, StateDispatching, now,
		digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StatePending, StateRetry, stateDependencyRegistration, now,
		StatePending, StateClaimed, StateDispatching, StateRetry, stateDependencyRegistration,
		stateWaitingDependency, StateWriteback, StateUncertain,
	).Scan(&found)
	if err != nil {
		return false, fmt.Errorf("connector outbound intent work hint: %w", err)
	}
	return found, nil
}

// Parameters in the CASE/COALESCE branches can make TiDB infer a string
// aggregate. Declare a temporal result so parseTime preserves microseconds.
const outboundNextWorkAtSQL = `
SELECT CAST(MIN(scheduled.WakeAt) AS DATETIME(6))
  FROM (
    SELECT CASE
             WHEN COALESCE(writeback.ClaimToken, '') <> '' AND writeback.LeaseExpiresAt > ?
               THEN writeback.LeaseExpiresAt
             ELSE COALESCE(writeback.NextAttemptAt, ?)
           END AS WakeAt
      FROM _System_IntegrationOutboundCommand AS writeback
     WHERE writeback.ScopeDigest = ?
       AND writeback.OrgId = ?
       AND writeback.AppInstallationId = ?
       AND writeback.AuthorizedWorkloadId = ?
       AND writeback.CatalogRevision = ?
       AND writeback.State = ?
    UNION ALL
    SELECT COALESCE(owned.LeaseExpiresAt, ?)
      FROM _System_IntegrationOutboundCommand AS owned
     WHERE owned.ScopeDigest = ?
       AND owned.OrgId = ?
       AND owned.AppInstallationId = ?
       AND owned.AuthorizedWorkloadId = ?
       AND owned.CatalogRevision = ?
       AND owned.State IN (?, ?)
    UNION ALL
    SELECT COALESCE(candidate.NextAttemptAt, ?)
      FROM _System_IntegrationOutboundCommand AS candidate
     WHERE candidate.ScopeDigest = ?
       AND candidate.OrgId = ?
       AND candidate.AppInstallationId = ?
       AND candidate.AuthorizedWorkloadId = ?
       AND candidate.CatalogRevision = ?
       AND candidate.State IN (?, ?, ?)
       AND NOT EXISTS (
         SELECT 1
           FROM _System_IntegrationOutboundCommand AS prior
          WHERE prior.ScopeDigest = candidate.ScopeDigest
            AND prior.OrgId = candidate.OrgId
            AND prior.AppInstallationId = candidate.AppInstallationId
            AND prior.AuthorizedWorkloadId = candidate.AuthorizedWorkloadId
            AND prior.CatalogRevision = candidate.CatalogRevision
            AND prior.LinkKey = candidate.LinkKey
            AND prior.State IN (?, ?, ?, ?, ?, ?, ?, ?)
            AND prior.AcceptanceSequence < candidate.AcceptanceSequence
       )
  ) AS scheduled`

// NextWorkAt returns the exact earliest durable retry, lease-recovery, or
// unblocked-lane wake time. It schedules a future authoritative claim; it does
// not authorize delivery and is never the queue source of truth.
func (r *Repository) NextWorkAt(ctx context.Context, scope Scope) (time.Time, bool, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil {
		return time.Time{}, false, ErrInvalidInput
	}
	now := r.now().UTC()
	digest := scopeDigest(scope)
	var next sql.NullTime
	err := r.db.QueryRowContext(
		ctx,
		outboundNextWorkAtSQL,
		now, now,
		digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StateWriteback,
		now,
		digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StateClaimed, StateDispatching,
		now,
		digest, scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StatePending, StateRetry, stateDependencyRegistration,
		StatePending, StateClaimed, StateDispatching, StateRetry, stateDependencyRegistration,
		stateWaitingDependency, StateWriteback, StateUncertain,
	).Scan(&next)
	if err != nil {
		return time.Time{}, false, fmt.Errorf("connector outbound intent next work time: %w", err)
	}
	if !next.Valid {
		return time.Time{}, false, nil
	}
	return next.Time.UTC(), true, nil
}

const outboundLeaseRecoverySQL = `
UPDATE _System_IntegrationOutboundCommand
	SET State = CASE
	       WHEN DispatchStartedAt IS NULL THEN ?
	       WHEN RemoteId IS NOT NULL AND FailureCode IN (?, ?) THEN ?
	       ELSE ?
	   END,
       ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL, LeaseExpiresAt = NULL,
	   FailureCode = CASE
	       WHEN DispatchStartedAt IS NULL OR
	            (RemoteId IS NOT NULL AND FailureCode IN (?, ?)) THEN FailureCode
	       ELSE 'dispatch_owner_lost'
	   END,
       LastModifiedDate = ?
 WHERE ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
   AND CatalogRevision = ? AND State IN (?, ?) AND LeaseExpiresAt <= ?`

const outboundClaimCandidateSQL = commandSelect + ` AS candidate
 WHERE candidate.ScopeDigest = ? AND candidate.OrgId = ? AND candidate.AppInstallationId = ? AND candidate.AuthorizedWorkloadId = ?
   AND candidate.CatalogRevision = ? AND candidate.State IN (?, ?, ?)
   AND (candidate.NextAttemptAt IS NULL OR candidate.NextAttemptAt <= ?)
   AND NOT EXISTS (
     SELECT 1 FROM _System_IntegrationOutboundCommand AS prior
      WHERE prior.ScopeDigest = candidate.ScopeDigest
		AND prior.OrgId = candidate.OrgId
        AND prior.AppInstallationId = candidate.AppInstallationId
        AND prior.AuthorizedWorkloadId = candidate.AuthorizedWorkloadId
        AND prior.CatalogRevision = candidate.CatalogRevision
        AND prior.LinkKey = candidate.LinkKey
        AND prior.State IN (?, ?, ?, ?, ?, ?, ?, ?)
		AND prior.AcceptanceSequence < candidate.AcceptanceSequence)
 ORDER BY COALESCE(candidate.NextAttemptAt, candidate.CreatedDate), candidate.AcceptanceSequence
 LIMIT 1 FOR UPDATE`

func (r *Repository) ClaimNext(ctx context.Context, scope Scope, workerID, token string, lease time.Duration) (Command, bool, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil ||
		strings.TrimSpace(workerID) == "" || workerID != strings.TrimSpace(workerID) ||
		strings.TrimSpace(token) == "" || token != strings.TrimSpace(token) || lease <= 0 {
		return Command{}, false, ErrInvalidInput
	}
	now := r.now().UTC()
	leaseUntil := now.Add(lease)
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, false, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, bool, error) {
		return Command{}, false, errors.Join(cause, tx.Rollback())
	}
	// A crashed claimant that had not crossed the dispatch checkpoint is safe
	// to reclaim. Once DispatchStartedAt is non-null, lease expiry normally
	// proves no provider outcome and is terminally isolated as uncertain. The
	// one exception is a checkpointed main success: that lane has a durable
	// remote identity and performs only idempotent association PUTs, so it is
	// safe to resume after owner loss.
	if _, err := tx.ExecContext(ctx, outboundLeaseRecoverySQL,
		StatePending,
		"association_create_outcome_uncertain", "association_update_outcome_uncertain", StateRetry,
		StateUncertain,
		"association_create_outcome_uncertain", "association_update_outcome_uncertain",
		now, scopeDigest(scope), scope.OrgID, scope.AppInstallationID,
		scope.AuthorizedWorkloadID, scope.CatalogRevision, StateClaimed, StateDispatching, now); err != nil {
		return rollback(err)
	}
	row := tx.QueryRowContext(ctx, outboundClaimCandidateSQL,
		scopeDigest(scope), scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StatePending, StateRetry, stateDependencyRegistration, now,
		StatePending, StateClaimed, StateDispatching, StateRetry, stateDependencyRegistration,
		stateWaitingDependency, StateWriteback, StateUncertain)
	command, err := scanCommand(row)
	if errors.Is(err, sql.ErrNoRows) {
		if err := tx.Commit(); err != nil {
			return Command{}, false, err
		}
		return Command{}, false, nil
	}
	if err != nil {
		return rollback(err)
	}
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
	SET State = ?, ClaimedBy = ?, ClaimToken = ?,
       ClaimedAt = ?, LeaseExpiresAt = ?, DispatchStartedAt = NULL,
       NextAttemptAt = NULL, DependencyKind = NULL, DependencyKey = NULL,
       DependencyReadinessKeys = NULL, DependencyBlockedAt = NULL, LastModifiedDate = ?
 WHERE Id = ? AND State IN (?, ?, ?) AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?)`,
		StateClaimed, workerID, token, now, leaseUntil, now, command.CommandID,
		StatePending, StateRetry, stateDependencyRegistration, now)
	if err != nil {
		return rollback(err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, false, err
	}
	command.State = StateClaimed
	command.ClaimedBy, command.ClaimToken = workerID, token
	command.ClaimedAt, command.LeaseExpiresAt = &now, &leaseUntil
	command.NextAttemptAt = nil
	return command, true, nil
}

func (r *Repository) ClaimWriteback(ctx context.Context, scope Scope, workerID, token string, lease time.Duration) (Command, bool, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil ||
		strings.TrimSpace(workerID) == "" || workerID != strings.TrimSpace(workerID) ||
		strings.TrimSpace(token) == "" || token != strings.TrimSpace(token) || lease <= 0 {
		return Command{}, false, ErrInvalidInput
	}
	now := r.now().UTC()
	leaseUntil := now.Add(lease)
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, false, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, bool, error) {
		return Command{}, false, errors.Join(cause, tx.Rollback())
	}
	row := tx.QueryRowContext(ctx, commandSelect+`
 WHERE ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
   AND CatalogRevision = ? AND State = ?
   AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?)
   AND (ClaimToken IS NULL OR ClaimToken = '' OR LeaseExpiresAt <= ?)
 ORDER BY COALESCE(NextAttemptAt, LastModifiedDate), Id
 LIMIT 1 FOR UPDATE`,
		scopeDigest(scope), scope.OrgID, scope.AppInstallationID, scope.AuthorizedWorkloadID,
		scope.CatalogRevision, StateWriteback, now, now)
	command, err := scanCommand(row)
	if errors.Is(err, sql.ErrNoRows) {
		if err := tx.Commit(); err != nil {
			return Command{}, false, err
		}
		return Command{}, false, nil
	}
	if err != nil {
		return rollback(err)
	}
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET ClaimedBy = ?, ClaimToken = ?,
       ClaimedAt = ?, LeaseExpiresAt = ?, NextAttemptAt = NULL,
       LastModifiedDate = ?
 WHERE Id = ? AND State = ?
   AND (ClaimToken IS NULL OR ClaimToken = '' OR LeaseExpiresAt <= ?)`,
		workerID, token, now, leaseUntil, now, command.CommandID, StateWriteback, now)
	if err != nil {
		return rollback(err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, false, err
	}
	command.ClaimedBy, command.ClaimToken = workerID, token
	command.ClaimedAt, command.LeaseExpiresAt = &now, &leaseUntil
	command.NextAttemptAt = nil
	return command, true, nil
}

func (r *Repository) MarkDispatching(ctx context.Context, claim Claim) (Command, error) {
	return r.markDispatchingWithPreparedProof(ctx, claim, nil, nil)
}

// MarkDispatchingPrepared consumes an opaque prepared link snapshot in the
// same transaction that admits provider dispatch.
func (r *Repository) MarkDispatchingPrepared(ctx context.Context, claim Claim, proof PreparedDispatchProof) (Command, error) {
	if proof.scoped == nil {
		return Command{}, ErrInvalidInput
	}
	return r.markDispatchingWithPreparedProof(ctx, claim, proof.scoped, proof.legacy)
}

func (r *Repository) markDispatchingWithLegacyProof(ctx context.Context, claim Claim, proof *preparedLegacyRemoteLink) (Command, error) {
	return r.markDispatchingWithPreparedProof(ctx, claim, nil, proof)
}

func (r *Repository) markDispatchingWithPreparedProof(ctx context.Context, claim Claim, scopedProof *preparedScopedRemoteLink, legacyProof *preparedLegacyRemoteLink) (Command, error) {
	if r == nil || r.db == nil || ctx == nil {
		return Command{}, ErrDependency
	}
	if err := validateClaim(claim); err != nil {
		return Command{}, err
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, error) {
		return Command{}, errors.Join(cause, tx.Rollback())
	}
	locked, err := scanCommand(tx.QueryRowContext(ctx, commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`, claim.Command.CommandID))
	if err != nil {
		return rollback(err)
	}
	if locked.State != StateClaimed || locked.ClaimToken != claim.Token ||
		locked.DispatchStarted != nil || locked.LeaseExpiresAt == nil ||
		!locked.LeaseExpiresAt.After(r.now().UTC()) ||
		!sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}
	if locked.Intent.Operation == outboundintentv1.OperationUpsert {
		if legacyProof != nil {
			if err := lockPreparedLegacyRemoteLink(ctx, tx, locked, legacyProof); err != nil {
				return rollback(err)
			}
			if err := rejectForeignLegacyCommands(ctx, tx, locked); err != nil {
				return rollback(err)
			}
		}
		if scopedProof != nil {
			if err := lockPreparedScopedRemoteLink(ctx, tx, locked, scopedProof); err != nil {
				return rollback(err)
			}
		} else {
			// Prepare read the link before this transaction. Serialize this last
			// pre-I/O checkpoint with identity transfer using the established
			// Command -> Link order. A transfer holding Link can read the committed
			// command state without locking it; it must reject an in-flight command.
			// If retirement wins first, no prepared request may cross this fence.
			link, found, err := lockRemoteLink(ctx, tx, locked.LinkKey)
			if err != nil {
				return rollback(err)
			}
			// Preparation chose V1 only because V2 was absent. A concurrent V2
			// publication requires preparing again, not silently retargeting PATCH.
			if legacyProof != nil && found {
				return rollback(newDurableIdentityConflict(locked, "dispatch_admission", "ConcurrentRemoteLink"))
			}
			if found && !validDispatchRemoteLink(link, locked) {
				return rollback(newDurableIdentityConflict(locked, "dispatch_admission", "RemoteLinkIdentity"))
			}
			// No existing link is the legitimate CREATE path. This fence does not
			// introduce an unscoped identity or treat a missing row as a gap lock.
		}
	}
	now := r.now().UTC()
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, AttemptCount = AttemptCount + 1,
       DispatchStartedAt = ?, LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ? AND DispatchStartedAt IS NULL`,
		StateDispatching, now, now, claim.Command.CommandID, StateClaimed, claim.Token, now)
	if err != nil {
		return rollback(err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return rollback(err)
	}
	if affected != 1 {
		return rollback(ErrLeaseLost)
	}
	return commitCommand(ctx, tx, locked.CommandID)
}

// Retry preserves the released scheduling seam while treating next as the
// caller's minimum. The repository owns the durable retry series and may move
// the command later to enforce capped backoff across process restarts.
func (r *Repository) Retry(ctx context.Context, claim Claim, next time.Time, code string) (Command, error) {
	if r == nil || r.db == nil || ctx == nil || validateClaim(claim) != nil || !safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	requestedAt := r.now().UTC()
	if !next.After(requestedAt) {
		return Command{}, ErrInvalidInput
	}
	minimum := next.Sub(requestedAt)
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, error) {
		return Command{}, errors.Join(cause, tx.Rollback())
	}
	locked, retryCount, err := scanStoredCommand(tx.QueryRowContext(
		ctx, commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`, claim.Command.CommandID,
	))
	if err != nil {
		return rollback(err)
	}
	now := r.now().UTC()
	if (locked.State != StateClaimed && locked.State != StateDispatching) ||
		locked.ClaimToken != claim.Token || locked.LeaseExpiresAt == nil ||
		!locked.LeaseExpiresAt.After(now) || !sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}
	retryAt := now.Add(safeRetryDelay(locked.CommandID, retryCount, minimum))
	persistedCode := code
	if locked.RemoteID != "" &&
		(locked.FailureCode == "association_create_outcome_uncertain" ||
			locked.FailureCode == "association_update_outcome_uncertain") {
		persistedCode = locked.FailureCode
	}
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, RetryCount = RetryCount + 1,
       ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, DispatchStartedAt = NULL, NextAttemptAt = ?,
       FailureCode = ?, LastModifiedDate = ?
 WHERE Id = ? AND State IN (?, ?) AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ?`,
		StateRetry, retryAt, persistedCode, now, locked.CommandID,
		StateClaimed, StateDispatching, claim.Token, now)
	if err != nil {
		return rollback(err)
	}
	if affected, affectedErr := result.RowsAffected(); affectedErr != nil || affected != 1 {
		return rollback(ErrLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	locked.State, locked.FailureCode = StateRetry, persistedCode
	locked.NextAttemptAt, locked.DispatchStarted = &retryAt, nil
	locked.ClaimedBy, locked.ClaimToken, locked.ClaimedAt, locked.LeaseExpiresAt = "", "", nil, nil
	locked.ModifiedAt = now
	return locked, nil
}

// RetryWriteback applies the same durable safe-retry policy to the idempotent
// Runtime projection lane without exposing a second store contract.
func (r *Repository) RetryWriteback(ctx context.Context, claim Claim, next time.Time, code string) (Command, error) {
	if r == nil || r.db == nil || ctx == nil || validateClaim(claim) != nil || !safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	requestedAt := r.now().UTC()
	if !next.After(requestedAt) {
		return Command{}, ErrInvalidInput
	}
	minimum := next.Sub(requestedAt)
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, error) {
		return Command{}, errors.Join(cause, tx.Rollback())
	}
	locked, retryCount, err := scanStoredCommand(tx.QueryRowContext(
		ctx, commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`, claim.Command.CommandID,
	))
	if err != nil {
		return rollback(err)
	}
	now := r.now().UTC()
	if locked.State != StateWriteback || locked.ClaimToken != claim.Token || locked.LeaseExpiresAt == nil ||
		!locked.LeaseExpiresAt.After(now) || !sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}
	retryAt := now.Add(safeRetryDelay(locked.CommandID, retryCount, minimum))
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET RetryCount = RetryCount + 1,
       ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, NextAttemptAt = ?, FailureCode = ?, LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ?`,
		retryAt, code, now, locked.CommandID, StateWriteback, claim.Token, now)
	if err != nil {
		return rollback(err)
	}
	if affected, affectedErr := result.RowsAffected(); affectedErr != nil || affected != 1 {
		return rollback(ErrLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	locked.FailureCode, locked.NextAttemptAt = code, &retryAt
	locked.ClaimedBy, locked.ClaimToken, locked.ClaimedAt, locked.LeaseExpiresAt = "", "", nil, nil
	locked.ModifiedAt = now
	return locked, nil
}

func (r *Repository) Reject(ctx context.Context, claim Claim, status int, code string) (Command, error) {
	return r.finish(ctx, claim, StateRejected, status, "", code)
}

func (r *Repository) RecordProviderSuccess(ctx context.Context, claim Claim, status int, finalization RemoteLinkFinalization) (Command, error) {
	return r.recordProviderSuccessWithLegacyProof(ctx, claim, status, finalization, nil)
}

func (r *Repository) recordProviderSuccessWithLegacyProof(ctx context.Context, claim Claim, status int, finalization RemoteLinkFinalization, proof *preparedLegacyRemoteLink) (Command, error) {
	if status < 200 || status >= 300 || validateRemoteLinkFinalization(claim, finalization) != nil {
		return Command{}, ErrInvalidInput
	}
	return r.succeedWithRemoteLink(ctx, claim, status, finalization, nil, proof)
}

type providerSuccessDisposition struct {
	state     string
	minimum   *time.Duration
	code      string
	keepClaim bool
}

func (r *Repository) CheckpointProviderSuccess(
	ctx context.Context,
	claim Claim,
	status int,
	finalization RemoteLinkFinalization,
	code string,
) (Command, error) {
	return r.checkpointProviderSuccessWithLegacyProof(ctx, claim, status, finalization, code, nil)
}

func (r *Repository) checkpointProviderSuccessWithLegacyProof(ctx context.Context, claim Claim, status int, finalization RemoteLinkFinalization, code string, proof *preparedLegacyRemoteLink) (Command, error) {
	if status < 200 || status >= 300 || validateRemoteLinkFinalization(claim, finalization) != nil ||
		!safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	return r.succeedWithRemoteLink(ctx, claim, status, finalization, &providerSuccessDisposition{
		state: StateDispatching, code: code, keepClaim: true,
	}, proof)
}

// RecordProviderSuccessForRetry checkpoints provider success and schedules
// only its safe association continuation through the same retry series.
func (r *Repository) RecordProviderSuccessForRetry(
	ctx context.Context,
	claim Claim,
	status int,
	finalization RemoteLinkFinalization,
	next time.Time,
	code string,
) (Command, error) {
	return r.recordProviderSuccessForRetryWithLegacyProof(ctx, claim, status, finalization, next, code, nil)
}

func (r *Repository) recordProviderSuccessForRetryWithLegacyProof(ctx context.Context, claim Claim, status int, finalization RemoteLinkFinalization, next time.Time, code string, proof *preparedLegacyRemoteLink) (Command, error) {
	if r == nil {
		return Command{}, ErrInvalidInput
	}
	now := r.now().UTC()
	if !next.After(now) {
		return Command{}, ErrInvalidInput
	}
	if status < 200 || status >= 300 || validateRemoteLinkFinalization(claim, finalization) != nil ||
		!safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	minimum := next.Sub(now)
	return r.succeedWithRemoteLink(ctx, claim, status, finalization, &providerSuccessDisposition{
		state: StateRetry, minimum: &minimum, code: code,
	}, proof)
}

func (r *Repository) RecordProviderSuccessTerminal(
	ctx context.Context,
	claim Claim,
	status int,
	finalization RemoteLinkFinalization,
	code string,
) (Command, error) {
	return r.recordProviderSuccessTerminalWithLegacyProof(ctx, claim, status, finalization, code, nil)
}

func (r *Repository) recordProviderSuccessTerminalWithLegacyProof(ctx context.Context, claim Claim, status int, finalization RemoteLinkFinalization, code string, proof *preparedLegacyRemoteLink) (Command, error) {
	if status < 200 || status >= 300 || validateRemoteLinkFinalization(claim, finalization) != nil || !safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	return r.succeedWithRemoteLink(ctx, claim, status, finalization, &providerSuccessDisposition{
		state: StateUncertain, code: code,
	}, proof)
}

// recordTypedResult persists a bounded terminal result on the existing
// Integration outbound command before Runtime writeback. It deliberately does
// not create or consult a second provider queue.
func (r *Repository) recordTypedResult(
	ctx context.Context,
	claim Claim,
	status int,
	result storedTypedTerminalResult,
) (Command, error) {
	if r == nil || r.db == nil || ctx == nil {
		return Command{}, ErrDependency
	}
	available, err := r.typedResultsAvailable(ctx)
	if err != nil {
		return Command{}, err
	}
	if !available {
		return Command{}, ErrSchemaUnverified
	}
	request := outboundintentv1.CommitResultRequest{
		IntentID: claim.Command.Intent.IntentID, PayloadDigest: claim.Command.Intent.PayloadDigest,
		CatalogRevision: claim.Command.Scope.CatalogRevision, IntegrationCommandID: claim.Command.CommandID,
		ResultFormatVersion: result.formatVersion, ResultDigest: result.digest,
		CanonicalResult: append([]byte(nil), result.canonical...),
	}
	if validateClaim(claim) != nil || (status != 0 && (status < 100 || status > 599)) ||
		outboundintentv1.ValidateCommitResultRequestForIntent(claim.Command.Intent, request) != nil {
		return Command{}, ErrInvalidInput
	}
	now := r.now().UTC()
	return r.transitionClaim(ctx, claim, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, DispatchStartedAt = NULL,
       NextAttemptAt = NULL, FailureCode = NULL, RetryCount = 0,
       ResponseStatus = NULLIF(?, 0), RemoteId = NULL,
       ResultFormatVersion = ?, ResultDigest = ?, CanonicalResult = ?,
       LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ? AND DispatchStartedAt IS NOT NULL`,
		StateWriteback, status, result.formatVersion, result.digest,
		append([]byte(nil), result.canonical...), now,
		claim.Command.CommandID, StateDispatching, claim.Token, now,
	)
}

func (r *Repository) loadTypedResult(ctx context.Context, claim Claim) (storedTypedTerminalResult, error) {
	if r == nil || r.db == nil || ctx == nil {
		return storedTypedTerminalResult{}, ErrDependency
	}
	if validateClaim(claim) != nil {
		return storedTypedTerminalResult{}, ErrInvalidInput
	}
	available, err := r.typedResultsAvailable(ctx)
	if err != nil {
		return storedTypedTerminalResult{}, err
	}
	if !available {
		return storedTypedTerminalResult{}, ErrSchemaUnverified
	}
	row := r.db.QueryRowContext(ctx, `
SELECT COALESCE(ResultFormatVersion, 0), COALESCE(ResultDigest, ''), COALESCE(CanonicalResult, '')
  FROM _System_IntegrationOutboundCommand
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ? AND LeaseExpiresAt > ?
 LIMIT 1`, claim.Command.CommandID, StateWriteback, claim.Token, r.now().UTC())
	return scanTypedTerminalResult(row, claim.Command)
}

func scanTypedTerminalResult(row scanner, command Command) (storedTypedTerminalResult, error) {
	var result storedTypedTerminalResult
	if err := row.Scan(&result.formatVersion, &result.digest, &result.canonical); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return storedTypedTerminalResult{}, ErrCommandNotFound
		}
		return storedTypedTerminalResult{}, err
	}
	if result.formatVersion == 0 {
		if result.digest != "" || len(result.canonical) != 0 {
			return storedTypedTerminalResult{}, ErrIdentityConflict
		}
		return storedTypedTerminalResult{}, ErrCommandNotFound
	}
	request := outboundintentv1.CommitResultRequest{
		IntentID: command.Intent.IntentID, PayloadDigest: command.Intent.PayloadDigest,
		CatalogRevision: command.Scope.CatalogRevision, IntegrationCommandID: command.CommandID,
		ResultFormatVersion: result.formatVersion, ResultDigest: result.digest,
		CanonicalResult: result.canonical,
	}
	if outboundintentv1.ValidateCommitResultRequestForIntent(command.Intent, request) != nil {
		return storedTypedTerminalResult{}, ErrIdentityConflict
	}
	return result, nil
}

func (r *Repository) Uncertain(ctx context.Context, claim Claim, status int, code string) (Command, error) {
	return r.finish(ctx, claim, StateUncertain, status, "", code)
}

// IsolatePostDispatch durably terminates one command after provider I/O may
// have occurred but its success or writeback identity could not be finalized.
// A successful isolation releases the claim without authorizing provider
// replay and preserves any provider evidence already recorded on the command.
func (r *Repository) IsolatePostDispatch(
	ctx context.Context,
	claim Claim,
	status int,
	remoteID string,
	code string,
) (Command, error) {
	if err := validateClaim(claim); err != nil ||
		(status != 0 && (status < 100 || status > 599)) ||
		remoteID != strings.TrimSpace(remoteID) || !safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	now := r.now().UTC()
	return r.transitionClaim(ctx, claim, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, NextAttemptAt = NULL, FailureCode = ?, RetryCount = 0,
       ResponseStatus = COALESCE(ResponseStatus, NULLIF(?, 0)),
       RemoteId = COALESCE(RemoteId, NULLIF(?, '')), LastModifiedDate = ?
 WHERE Id = ? AND State IN (?, ?) AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ?
   AND ((State = ? AND DispatchStartedAt IS NOT NULL) OR State = ?)
   AND (ResponseStatus IS NULL OR ? = 0 OR ResponseStatus = ?)
   AND (RemoteId IS NULL OR ? = '' OR BINARY RemoteId = BINARY ?)`,
		StateUncertain, code, status, remoteID, now,
		claim.Command.CommandID, StateDispatching, StateWriteback, claim.Token, now,
		StateDispatching, StateWriteback,
		status, status, remoteID, remoteID,
	)
}

func (r *Repository) finish(ctx context.Context, claim Claim, state string, status int, remoteID, code string) (Command, error) {
	if err := validateClaim(claim); err != nil || (status != 0 && (status < 100 || status > 599)) ||
		(state != StateSucceeded && !safeCode(code)) {
		return Command{}, ErrInvalidInput
	}
	now := r.now().UTC()
	if state == StateRejected {
		return r.transitionClaim(ctx, claim, `
UPDATE _System_IntegrationOutboundCommand
   SET State = CASE WHEN RemoteId IS NULL THEN ? ELSE ? END, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, NextAttemptAt = NULL, FailureCode = NULLIF(?, ''), RetryCount = 0,
       ResponseStatus = COALESCE(ResponseStatus, NULLIF(?, 0)), RemoteId = COALESCE(RemoteId, NULLIF(?, '')), LastModifiedDate = ?
	 WHERE Id = ? AND State IN (?, ?) AND BINARY ClaimToken = BINARY ?
	   AND LeaseExpiresAt > ?
	   AND ((State = ? AND DispatchStartedAt IS NULL) OR
	        (State = ? AND DispatchStartedAt IS NOT NULL))`,
			state, StateUncertain, code, status, remoteID, now, claim.Command.CommandID,
			StateClaimed, StateDispatching, claim.Token, now,
			StateClaimed, StateDispatching)
	}
	return r.transitionClaim(ctx, claim, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, NextAttemptAt = NULL, FailureCode = NULLIF(?, ''), RetryCount = 0,
       ResponseStatus = NULLIF(?, 0), RemoteId = NULLIF(?, ''), LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ? AND DispatchStartedAt IS NOT NULL`,
		state, code, status, remoteID, now, claim.Command.CommandID,
		StateDispatching, claim.Token, now)
}

func (r *Repository) transitionClaim(ctx context.Context, claim Claim, query string, args ...any) (Command, error) {
	if r == nil || r.db == nil || ctx == nil {
		return Command{}, ErrDependency
	}
	result, err := r.db.ExecContext(ctx, query, args...)
	if err != nil {
		return Command{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return Command{}, ErrLeaseLost
	}
	return r.loadByID(ctx, claim.Command.CommandID)
}

func validateClaim(claim Claim) error {
	if claim.Command.CommandID == "" || claim.Token == "" || claim.Token != strings.TrimSpace(claim.Token) ||
		claim.Command.ClaimToken != claim.Token {
		return ErrInvalidInput
	}
	return nil
}

func safeCode(value string) bool {
	if value == "" || len(value) > 64 || value != strings.TrimSpace(value) {
		return false
	}
	for _, current := range value {
		if !((current >= 'a' && current <= 'z') || (current >= '0' && current <= '9') || current == '_' || current == '.') {
			return false
		}
	}
	return true
}

const commandSelect = `SELECT
 Id, AcceptanceSequence, OrgId, AppInstallationId, AuthorizedWorkloadId, ScopeDigest, CatalogRevision,
 IntentId, LinkKey, IdempotencyKey, PayloadFormatVersion, PayloadDigest, Operation,
 SourceObject, SourceRecordId, ConnectorOwnerKey, FlowDeveloperName,
 FlowActionName, ActionSlotId, DestinationSlotId, Bindings, CanonicalPayload,
 State, AttemptCount, RetryCount, COALESCE(ClaimedBy, ''), COALESCE(ClaimToken, ''),
 ClaimedAt, LeaseExpiresAt, DispatchStartedAt, NextAttemptAt,
 COALESCE(FailureCode, ''), COALESCE(ResponseStatus, 0), COALESCE(RemoteId, ''),
 CreatedDate, LastModifiedDate
 FROM _System_IntegrationOutboundCommand`

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

func scanCommand(row scanner) (Command, error) {
	command, _, err := scanStoredCommand(row)
	return command, err
}

func scanStoredCommand(row scanner) (Command, int, error) {
	var command Command
	var retryCount int
	var operation string
	var bindings []byte
	var persistedScopeDigest string
	var claimedAt, lease, dispatchedAt, next sql.NullTime
	if err := row.Scan(
		&command.CommandID, &command.AcceptanceSequence, &command.Scope.OrgID, &command.Scope.AppInstallationID,
		&command.Scope.AuthorizedWorkloadID, &persistedScopeDigest, &command.Scope.CatalogRevision,
		&command.Intent.IntentID, &command.LinkKey, &command.Intent.IdempotencyKey,
		&command.Intent.PayloadFormatVersion, &command.Intent.PayloadDigest, &operation,
		&command.Intent.SourceObject, &command.Intent.SourceRecordID,
		&command.Intent.ConnectorOwnerKey, &command.Intent.FlowDeveloperName,
		&command.Intent.FlowActionName, &command.Intent.ActionSlotID,
		&command.Intent.DestinationSlotID, &bindings, &command.Intent.CanonicalPayload,
		&command.State, &command.AttemptCount, &retryCount, &command.ClaimedBy, &command.ClaimToken,
		&claimedAt, &lease, &dispatchedAt, &next, &command.FailureCode,
		&command.ResponseStatus, &command.RemoteID, &command.CreatedAt, &command.ModifiedAt,
	); err != nil {
		return Command{}, 0, err
	}
	command.Intent.Operation = outboundintentv1.Operation(operation)
	command.Intent.CatalogRevision = command.Scope.CatalogRevision
	if persistedScopeDigest != scopeDigest(command.Scope) ||
		command.AttemptCount < 0 || retryCount < 0 ||
		json.Unmarshal(bindings, &command.Intent.Bindings) != nil ||
		outboundintentv1.ValidateIntent(command.Intent) != nil || command.Scope.validate() != nil {
		return Command{}, 0, ErrIdentityConflict
	}
	if claimedAt.Valid {
		command.ClaimedAt = &claimedAt.Time
	}
	if lease.Valid {
		command.LeaseExpiresAt = &lease.Time
	}
	if dispatchedAt.Valid {
		command.DispatchStarted = &dispatchedAt.Time
	}
	if next.Valid {
		command.NextAttemptAt = &next.Time
	}
	return command, retryCount, nil
}

func (r *Repository) loadByID(ctx context.Context, id string) (Command, error) {
	command, err := scanCommand(r.db.QueryRowContext(ctx, commandSelect+` WHERE Id = ? LIMIT 1`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return Command{}, ErrCommandNotFound
	}
	return command, err
}

func (r *Repository) loadByIdentity(ctx context.Context, scope Scope, intentID, idempotencyKey string) (Command, error) {
	return loadByIdentityFrom(ctx, r.db, scope, intentID, idempotencyKey)
}

type rowQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

func loadByIdentityFrom(ctx context.Context, db rowQuerier, scope Scope, intentID, idempotencyKey string) (Command, error) {
	return loadByIdentityWithSuffixFrom(ctx, db, scope, intentID, idempotencyKey, "")
}

func loadByIdentityForUpdateFrom(ctx context.Context, tx *sql.Tx, scope Scope, intentID, idempotencyKey string) (Command, error) {
	return loadByIdentityWithSuffixFrom(ctx, tx, scope, intentID, idempotencyKey, " FOR UPDATE")
}

func loadByIdentityWithSuffixFrom(ctx context.Context, db rowQuerier, scope Scope, intentID, idempotencyKey, suffix string) (Command, error) {
	command, err := loadByExactIdentityWithSuffixFrom(ctx, db, scope, "IntentId", intentID, suffix)
	if err == nil || idempotencyKey == "" || !errors.Is(err, ErrCommandNotFound) {
		return command, err
	}
	return loadByExactIdentityWithSuffixFrom(ctx, db, scope, "IdempotencyKey", idempotencyKey, suffix)
}

func loadByExactIdentityWithSuffixFrom(
	ctx context.Context,
	db rowQuerier,
	scope Scope,
	column string,
	value string,
	suffix string,
) (Command, error) {
	if (column != "IntentId" && column != "IdempotencyKey") || value == "" {
		return Command{}, ErrCommandNotFound
	}
	command, err := scanCommand(db.QueryRowContext(ctx, commandSelect+`
 WHERE ScopeDigest = ? AND `+column+` = ?
   AND OrgId = ? AND AppInstallationId = ? AND AuthorizedWorkloadId = ?
   AND CatalogRevision = ?
 LIMIT 1`+suffix, scopeDigest(scope), value, scope.OrgID, scope.AppInstallationID,
		scope.AuthorizedWorkloadID, scope.CatalogRevision))
	if errors.Is(err, sql.ErrNoRows) {
		return Command{}, ErrCommandNotFound
	}
	if err != nil {
		return Command{}, fmt.Errorf("connector outbound intent load by %s: %w", column, err)
	}
	return command, nil
}

// commitCommand reads the updated row through the transaction before committing.
// A pessimistic transaction keeps its dedicated connection until cleanup; a
// second pool read here can deadlock when the pool is intentionally size one.
func commitCommand(ctx context.Context, tx *sql.Tx, id string) (Command, error) {
	command, err := scanCommand(tx.QueryRowContext(ctx, commandSelect+` WHERE Id = ? LIMIT 1`, id))
	if err != nil {
		return Command{}, errors.Join(err, tx.Rollback())
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	return command, nil
}

func sameCommand(command Command, scope Scope, intent outboundintentv1.Intent) bool {
	return command.CommandID == commandID(scope, intent) && command.Scope == scope &&
		command.LinkKey == outboundLinkKey(scope, intent) &&
		command.Intent.IntentID == intent.IntentID && command.Intent.IdempotencyKey == intent.IdempotencyKey &&
		command.Intent.PayloadFormatVersion == intent.PayloadFormatVersion &&
		command.Intent.PayloadDigest == intent.PayloadDigest && command.Intent.Operation == intent.Operation &&
		command.Intent.SourceObject == intent.SourceObject && command.Intent.SourceRecordID == intent.SourceRecordID &&
		command.Intent.ConnectorOwnerKey == intent.ConnectorOwnerKey &&
		command.Intent.FlowDeveloperName == intent.FlowDeveloperName &&
		command.Intent.FlowActionName == intent.FlowActionName && command.Intent.ActionSlotID == intent.ActionSlotID &&
		command.Intent.DestinationSlotID == intent.DestinationSlotID &&
		reflect.DeepEqual(command.Intent.Bindings, intent.Bindings) &&
		bytes.Equal(command.Intent.CanonicalPayload, intent.CanonicalPayload)
}