package connector_workqueue

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"strings"
	"time"

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

const (
	maxDependencyReconcileBatch                 = 1000
	identityConvergenceProbeRetryDelay          = time.Minute
	identityConvergenceDependencyKindSQL        = "'" + contractmodels.FlowDependencyKindIdentityConvergence + "'"
	identityConvergenceInvalidPersistedStateSQL = `(e.Id IS NULL
        OR w.DependencyBlockedAt IS NULL
        OR TRIM(COALESCE(w.InstalledPackId, '')) = ''
        OR TRIM(COALESCE(e.Namespace, '')) = ''
        OR TRIM(COALESCE(e.SourceObjectKind, '')) = ''
        OR TRIM(COALESCE(e.SourceObjectID, '')) = ''
        OR TRIM(COALESCE(w.DependencyObjectAPIName, '')) = ''
        OR TRIM(COALESCE(w.DependencyRecordId, '')) = ''
        OR TRIM(COALESCE(w.DependencyKeyHash, '')) = ''
        OR TRIM(COALESCE(w.DependencyConnectorOwnerKey, '')) = ''
        OR TRIM(COALESCE(w.DependencyConnectorNamespace, '')) = ''
        OR TRIM(COALESCE(w.DependencyConnectorSourceObject, '')) = ''
        OR TRIM(COALESCE(w.DependencyDestinationSlotId, '')) = ''
        OR TRIM(COALESCE(w.DependencyConnectorRemoteId, '')) = ''
        OR COALESCE(w.DependencyObjectAPIName, '') <> TRIM(COALESCE(w.DependencyObjectAPIName, ''))
        OR COALESCE(w.DependencyRecordId, '') <> TRIM(COALESCE(w.DependencyRecordId, ''))
        OR COALESCE(w.DependencyKeyHash, '') NOT REGEXP '^[0-9a-f]{64}$'
        OR COALESCE(w.DependencyConnectorOwnerKey, '') <> TRIM(COALESCE(w.DependencyConnectorOwnerKey, ''))
        OR COALESCE(w.DependencyConnectorNamespace, '') <> TRIM(COALESCE(w.DependencyConnectorNamespace, ''))
        OR COALESCE(w.DependencyConnectorSourceObject, '') <> TRIM(COALESCE(w.DependencyConnectorSourceObject, ''))
        OR COALESCE(w.DependencyDestinationSlotId, '') <> TRIM(COALESCE(w.DependencyDestinationSlotId, ''))
        OR COALESCE(w.DependencyConnectorRemoteId, '') <> TRIM(COALESCE(w.DependencyConnectorRemoteId, ''))
        OR w.DependencyConnectorNamespace <> e.Namespace
        OR w.DependencyConnectorSourceObject <> e.SourceObjectKind
        OR w.DependencyConnectorRemoteId <> e.SourceObjectID)`
)

// ExpireIdentityConvergenceWait terminalizes one exact lease-free dependency
// wait after its bounded convergence window. It first claims the observed
// dependency tuple, clears the active dependency coordinates, and then uses
// the normal failure transition so source-projection generations and lane
// promotion retain their existing single-owner semantics.
func (r *Repository) ExpireIdentityConvergenceWait(ctx context.Context, wait IdentityConvergenceWait, now time.Time) (bool, error) {
	if r == nil || r.db == nil || r.idgen == nil ||
		strings.TrimSpace(wait.WorkID) == "" ||
		wait.DependencyBlockedAt.IsZero() ||
		wait.Dependency.Validate() != nil ||
		now.IsZero() {
		return false, ErrInvalidWorkItem
	}

	q := r.db
	var tx *sql.Tx
	var cleanup func() error
	if db, ok := r.db.(*sql.DB); ok {
		var err error
		tx, cleanup, err = sqltx.BeginPessimisticTx(ctx, db)
		if err != nil {
			return false, err
		}
		defer cleanup()
		defer tx.Rollback()
		q = tx
	}

	claimToken := "identity-convergence-expiry:" + r.idgen()
	var observedProbeAfter any
	if wait.DependencyProbeAfter != nil {
		observedProbeAfter = *wait.DependencyProbeAfter
	}
	res, err := q.ExecContext(ctx, `
	UPDATE _System_ConnectorWorkItem
   SET LeaseOwner = ?,
       LeaseToken = ?,
       LeaseUntil = ?,
       DependencyKind = NULL,
       DependencyObjectAPIName = NULL,
       DependencyRecordId = NULL,
       DependencyKeyHash = NULL,
       DependencyConnectorOwnerKey = NULL,
       DependencyConnectorNamespace = NULL,
       DependencyConnectorSourceObject = NULL,
       DependencyDestinationSlotId = NULL,
       DependencyConnectorRemoteId = NULL,
       DependencyBlockedAt = NULL,
       DependencyProbeAfter = NULL,
       LastModifiedDate = ?
	 WHERE WorkId = ?
	   AND DependencyKind = ?
	   AND DependencyKeyHash = ?
	   AND DependencyBlockedAt = ?
	   AND DependencyProbeAfter <=> ?
	   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		"identity-convergence-expirer",
		claimToken,
		now.Add(time.Minute),
		now,
		strings.TrimSpace(wait.WorkID),
		contractmodels.FlowDependencyKindIdentityConvergence,
		strings.TrimSpace(wait.Dependency.KeyHash),
		wait.DependencyBlockedAt,
		observedProbeAfter,
		WorkStateProcessing,
	)
	if err != nil {
		return false, err
	}
	claimed, err := rowsAffected(res)
	if err != nil || !claimed {
		return claimed, err
	}

	var eventID string
	if err := q.QueryRowContext(ctx, `
SELECT EventId
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?
 LIMIT 1`, strings.TrimSpace(wait.WorkID), WorkStateProcessing, claimToken).Scan(&eventID); err != nil {
		return false, err
	}

	transitionRepo := &Repository{db: q, idgen: r.idgen, now: r.now}
	result, completed, err := transitionRepo.FailWorkItem(ctx, FailWorkItemInput{
		WorkID:         strings.TrimSpace(wait.WorkID),
		EventID:        strings.TrimSpace(eventID),
		LeaseToken:     claimToken,
		FailureClass:   FailureClassIdentityConvergenceExpired,
		FailureSummary: FailureSummaryIdentityConvergenceExpired,
		RetryThreshold: 1,
		RetryDelays:    []time.Duration{0},
		Now:            now,
	})
	if err != nil {
		return false, err
	}
	if !completed || result == nil || result.State != WorkStateDead {
		return false, fmt.Errorf("connector workqueue: identity convergence expiry did not reach dead terminal state")
	}
	if tx != nil {
		if err := tx.Commit(); err != nil {
			return false, err
		}
	}
	return true, nil
}

// TerminalizeInvalidIdentityConvergenceWait isolates one malformed durable wait
// at the exact observed row version. This prevents one corrupt dependency or
// missing Event provenance row from poisoning every unrelated reconciliation
// page while preserving the normal WorkItem dead transition and lane promotion.
func (r *Repository) TerminalizeInvalidIdentityConvergenceWait(
	ctx context.Context,
	workID string,
	now time.Time,
) (bool, error) {
	if r == nil || r.db == nil || r.idgen == nil ||
		strings.TrimSpace(workID) == "" ||
		now.IsZero() {
		return false, ErrInvalidWorkItem
	}
	workID = strings.TrimSpace(workID)

	q := r.db
	var tx *sql.Tx
	var cleanup func() error
	if db, ok := r.db.(*sql.DB); ok {
		var err error
		tx, cleanup, err = sqltx.BeginPessimisticTx(ctx, db)
		if err != nil {
			return false, err
		}
		defer cleanup()
		defer tx.Rollback()
		q = tx
	}

	var (
		installedPackID          sql.NullString
		dependencyBlockedAt      sql.NullTime
		namespace                sql.NullString
		objectKind               sql.NullString
		sourceID                 sql.NullString
		dependencyKind           sql.NullString
		dependencyObjectAPIName  sql.NullString
		dependencyRecordID       sql.NullString
		dependencyKeyHash        sql.NullString
		dependencyOwnerKey       sql.NullString
		dependencyNamespace      sql.NullString
		dependencySourceObject   sql.NullString
		dependencyDestinationID  sql.NullString
		dependencyRemoteID       sql.NullString
		observedLastModifiedDate sql.NullTime
	)
	err := q.QueryRowContext(ctx, `
SELECT w.InstalledPackId,
       w.DependencyBlockedAt,
       e.Namespace,
       e.SourceObjectKind,
       e.SourceObjectID,
       w.DependencyKind,
       w.DependencyObjectAPIName,
       w.DependencyRecordId,
       w.DependencyKeyHash,
       w.DependencyConnectorOwnerKey,
       w.DependencyConnectorNamespace,
       w.DependencyConnectorSourceObject,
       w.DependencyDestinationSlotId,
       w.DependencyConnectorRemoteId,
       w.LastModifiedDate
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.WorkId = ?
   AND w.DependencyKind = ?
   AND w.State = ?
   AND w.LeaseOwner IS NULL
   AND w.LeaseToken IS NULL
   AND w.LeaseUntil IS NULL
 LIMIT 1
 FOR UPDATE`, workID, contractmodels.FlowDependencyKindIdentityConvergence, WorkStateProcessing).Scan(
		&installedPackID,
		&dependencyBlockedAt,
		&namespace,
		&objectKind,
		&sourceID,
		&dependencyKind,
		&dependencyObjectAPIName,
		&dependencyRecordID,
		&dependencyKeyHash,
		&dependencyOwnerKey,
		&dependencyNamespace,
		&dependencySourceObject,
		&dependencyDestinationID,
		&dependencyRemoteID,
		&observedLastModifiedDate,
	)
	if err != nil {
		if err == sql.ErrNoRows {
			return false, nil
		}
		return false, err
	}
	wait := IdentityConvergenceWait{
		WorkID:              workID,
		InstalledPackID:     installedPackID.String,
		DependencyBlockedAt: dependencyBlockedAt.Time,
		Namespace:           namespace.String,
		ObjectKind:          objectKind.String,
		SourceID:            sourceID.String,
		Dependency: contractmodels.FlowExecutionDependency{
			Kind:                  dependencyKind.String,
			ObjectAPIName:         dependencyObjectAPIName.String,
			RecordID:              dependencyRecordID.String,
			KeyHash:               dependencyKeyHash.String,
			ConnectorOwnerKey:     dependencyOwnerKey.String,
			ConnectorNamespace:    dependencyNamespace.String,
			ConnectorSourceObject: dependencySourceObject.String,
			DestinationSlotID:     dependencyDestinationID.String,
			ConnectorRemoteID:     dependencyRemoteID.String,
		},
	}
	if !IsIdentityConvergenceWaitInvalid(wait) {
		return false, nil
	}
	if !observedLastModifiedDate.Valid || observedLastModifiedDate.Time.IsZero() {
		return false, fmt.Errorf("connector workqueue: invalid dependency row has no claimable LastModifiedDate")
	}

	claimToken := "identity-convergence-invalid:" + r.idgen()
	res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET LeaseOwner = ?,
       LeaseToken = ?,
       LeaseUntil = ?,
       DependencyKind = NULL,
       DependencyObjectAPIName = NULL,
       DependencyRecordId = NULL,
       DependencyKeyHash = NULL,
       DependencyConnectorOwnerKey = NULL,
       DependencyConnectorNamespace = NULL,
       DependencyConnectorSourceObject = NULL,
       DependencyDestinationSlotId = NULL,
       DependencyConnectorRemoteId = NULL,
       DependencyBlockedAt = NULL,
       DependencyProbeAfter = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND LastModifiedDate = ?
   AND DependencyKind = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		"identity-convergence-invalid-terminalizer",
		claimToken,
		now.Add(time.Minute),
		now,
		workID,
		observedLastModifiedDate.Time,
		contractmodels.FlowDependencyKindIdentityConvergence,
		WorkStateProcessing,
	)
	if err != nil {
		return false, err
	}
	claimed, err := rowsAffected(res)
	if err != nil || !claimed {
		return claimed, err
	}

	var eventID string
	if err := q.QueryRowContext(ctx, `
SELECT EventId
  FROM _System_ConnectorWorkItem
 WHERE WorkId = ?
   AND State = ?
   AND LeaseToken = ?
 LIMIT 1`, workID, WorkStateProcessing, claimToken).Scan(&eventID); err != nil {
		return false, err
	}

	transitionRepo := &Repository{db: q, idgen: r.idgen, now: r.now}
	result, completed, err := transitionRepo.FailWorkItem(ctx, FailWorkItemInput{
		WorkID:         workID,
		EventID:        strings.TrimSpace(eventID),
		LeaseToken:     claimToken,
		FailureClass:   FailureClassIdentityConvergenceInvalid,
		FailureSummary: FailureSummaryIdentityConvergenceInvalid,
		RetryThreshold: 1,
		RetryDelays:    []time.Duration{0},
		Now:            now,
	})
	if err != nil {
		return false, err
	}
	if !completed || result == nil || result.State != WorkStateDead {
		return false, fmt.Errorf("connector workqueue: invalid identity convergence wait did not reach dead terminal state")
	}
	if tx != nil {
		if err := tx.Commit(); err != nil {
			return false, err
		}
	}
	return true, nil
}

// ListInvalidIdentityConvergenceWaitIDs returns only structurally invalid
// dependency rows, including rows whose immutable Event provenance is missing.
// Keeping this sweep separate leaves the normal due/restart query shape intact.
func (r *Repository) ListInvalidIdentityConvergenceWaitIDs(
	ctx context.Context,
	limit int,
) ([]string, error) {
	if r == nil || r.db == nil || limit <= 0 || limit > maxDependencyReconcileBatch {
		return nil, fmt.Errorf("connector workqueue: invalid dependency isolation limit must be between 1 and %d", maxDependencyReconcileBatch)
	}
	rows, err := r.db.QueryContext(ctx, `
SELECT w.WorkId
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.State = ?
   AND w.DependencyKind = `+identityConvergenceDependencyKindSQL+`
   AND w.LeaseOwner IS NULL
   AND w.LeaseToken IS NULL
   AND w.LeaseUntil IS NULL
   AND `+identityConvergenceInvalidPersistedStateSQL+`
 ORDER BY w.WorkId ASC
 LIMIT ?`, WorkStateProcessing, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	workIDs := make([]string, 0)
	for rows.Next() {
		var workID string
		if err := rows.Scan(&workID); err != nil {
			return nil, err
		}
		workID = strings.TrimSpace(workID)
		if workID == "" {
			return nil, fmt.Errorf("connector workqueue: invalid dependency isolation row is not claimable")
		}
		workIDs = append(workIDs, workID)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return workIDs, nil
}

// IsIdentityConvergenceWaitInvalid is the single Go validation path for a
// durable dependency row assembled from WorkItem state and Event provenance.
// It delegates KeyHash validation to the neutral contract rather than
// duplicating the length-prefixed hash algorithm in SQL.
func IsIdentityConvergenceWaitInvalid(wait IdentityConvergenceWait) bool {
	return strings.TrimSpace(wait.WorkID) == "" ||
		strings.TrimSpace(wait.InstalledPackID) == "" ||
		wait.DependencyBlockedAt.IsZero() ||
		strings.TrimSpace(wait.Namespace) == "" ||
		strings.TrimSpace(wait.ObjectKind) == "" ||
		strings.TrimSpace(wait.SourceID) == "" ||
		wait.Dependency.ConnectorNamespace != strings.TrimSpace(wait.Namespace) ||
		wait.Dependency.ConnectorSourceObject != strings.TrimSpace(wait.ObjectKind) ||
		wait.Dependency.ConnectorRemoteID != strings.TrimSpace(wait.SourceID) ||
		wait.Dependency.Validate() != nil
}

// BlockWorkItem atomically surrenders a live processing lease and persists an
// identity-convergence prerequisite without consuming FailureCount. It retains
// processing as the row's compatibility state but clears every lease field:
// legacy workers therefore keep its unconditional lane fence yet cannot claim
// or stale-reclaim it. The non-null dependency coordinates are the durable
// suspension authority for new workers and the exact producer wake paths.
func (r *Repository) BlockWorkItem(ctx context.Context, in BlockWorkItemInput) (bool, error) {
	if strings.TrimSpace(in.WorkID) == "" ||
		strings.TrimSpace(in.EventID) == "" ||
		strings.TrimSpace(in.LeaseToken) == "" ||
		in.Now.IsZero() ||
		in.Dependency.Validate() != nil {
		return false, ErrInvalidWorkItem
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       DependencyKind = ?,
       DependencyObjectAPIName = ?,
       DependencyRecordId = ?,
       DependencyKeyHash = ?,
       DependencyConnectorOwnerKey = ?,
       DependencyConnectorNamespace = ?,
       DependencyConnectorSourceObject = ?,
       DependencyDestinationSlotId = ?,
       DependencyConnectorRemoteId = ?,
       DependencyBlockedAt = ?,
	   DependencyProbeAfter = ?,
       LeaseOwner = NULL,
       LeaseToken = NULL,
       LeaseUntil = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND EventId = ?
   AND State = ?
   AND LeaseToken = ?`,
		WorkStateProcessing,
		in.Dependency.Kind,
		in.Dependency.ObjectAPIName,
		in.Dependency.RecordID,
		in.Dependency.KeyHash,
		in.Dependency.ConnectorOwnerKey,
		in.Dependency.ConnectorNamespace,
		in.Dependency.ConnectorSourceObject,
		in.Dependency.DestinationSlotID,
		in.Dependency.ConnectorRemoteID,
		in.Now,
		in.Now,
		in.Now,
		strings.TrimSpace(in.WorkID),
		strings.TrimSpace(in.EventID),
		WorkStateProcessing,
		strings.TrimSpace(in.LeaseToken),
	)
	if err != nil {
		return false, err
	}
	return rowsAffected(res)
}

// WakeRecordDependency requests re-evaluation after an exact record candidate
// commits. Every connector-scoped dependency blocked on that same local object
// and record selector transitions; repeated signals and terminal/dead rows are
// no-ops.
func (r *Repository) WakeRecordDependency(ctx context.Context, objectAPIName, recordID string, now time.Time) (int64, error) {
	objectAPIName = strings.TrimSpace(objectAPIName)
	recordID = strings.TrimSpace(recordID)
	if now.IsZero() || objectAPIName == "" || recordID == "" {
		return 0, ErrInvalidWorkItem
	}
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       DependencyKind = NULL,
       DependencyObjectAPIName = NULL,
       DependencyRecordId = NULL,
       DependencyKeyHash = NULL,
       DependencyConnectorOwnerKey = NULL,
       DependencyConnectorNamespace = NULL,
       DependencyConnectorSourceObject = NULL,
       DependencyDestinationSlotId = NULL,
       DependencyConnectorRemoteId = NULL,
       DependencyBlockedAt = NULL,
	   DependencyProbeAfter = NULL,
       LastModifiedDate = ?
 WHERE DependencyKind = ?
   AND DependencyObjectAPIName = ?
   AND DependencyRecordId = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		WorkStatePending,
		now,
		now,
		contractmodels.FlowDependencyKindIdentityConvergence,
		objectAPIName,
		recordID,
		WorkStateProcessing,
	)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// WakeRemoteIdentity requests re-evaluation after finalized connector identity
// publication for the durable authority tuple. ConnectorSourceObject is Event
// provenance owned by each waiter, not part of the finalized-link authority
// key, so one committed authority fact intentionally wakes every source-object
// waiter for that tuple. The signal never assigns an owner; each Flow must
// revalidate its full persisted Event scope and durable authority after reclaim.
func (r *Repository) WakeRemoteIdentity(ctx context.Context, signal IdentityConvergenceRemoteSignal, now time.Time) (int64, error) {
	signal.ConnectorOwnerKey = strings.TrimSpace(signal.ConnectorOwnerKey)
	signal.ConnectorNamespace = strings.TrimSpace(signal.ConnectorNamespace)
	signal.DestinationSlotID = strings.TrimSpace(signal.DestinationSlotID)
	signal.ConnectorRemoteID = strings.TrimSpace(signal.ConnectorRemoteID)
	if now.IsZero() ||
		signal.ConnectorOwnerKey == "" ||
		signal.ConnectorNamespace == "" ||
		signal.DestinationSlotID == "" ||
		signal.ConnectorRemoteID == "" {
		return 0, ErrInvalidWorkItem
	}
	// DependencyConnectorSourceObject is waiter-owned Event provenance. The
	// finalized-link producer cannot author it, and RemoteLocatorActiveKey does
	// not contain it, so it must not narrow this authority-tuple notification.
	res, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET State = ?,
       RunAfter = ?,
       DependencyKind = NULL,
       DependencyObjectAPIName = NULL,
       DependencyRecordId = NULL,
       DependencyKeyHash = NULL,
       DependencyConnectorOwnerKey = NULL,
       DependencyConnectorNamespace = NULL,
       DependencyConnectorSourceObject = NULL,
       DependencyDestinationSlotId = NULL,
       DependencyConnectorRemoteId = NULL,
       DependencyBlockedAt = NULL,
	   DependencyProbeAfter = NULL,
       LastModifiedDate = ?
 WHERE DependencyKind = ?
   AND DependencyConnectorOwnerKey = ?
   AND DependencyConnectorNamespace = ?
   AND DependencyDestinationSlotId = ?
   AND DependencyConnectorRemoteId = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		WorkStatePending,
		now,
		now,
		contractmodels.FlowDependencyKindIdentityConvergence,
		signal.ConnectorOwnerKey,
		signal.ConnectorNamespace,
		signal.DestinationSlotID,
		signal.ConnectorRemoteID,
		WorkStateProcessing,
	)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// ScheduleIdentityConvergenceProbeRetry records a bounded, non-business retry
// after an authority probe failed. It does not change WorkItem state or
// FailureCount and cannot reopen terminal work.
func (r *Repository) ScheduleIdentityConvergenceProbeRetry(ctx context.Context, workID, dependencyKeyHash string, now time.Time) error {
	return r.scheduleIdentityConvergenceProbe(ctx, workID, dependencyKeyHash, now.Add(identityConvergenceProbeRetryDelay), now)
}

// RescheduleIdentityConvergenceProbeRetry advances only the exact due wait
// observed by a supervisor probe. The deadline comparison prevents a stale
// probe from overwriting a newer wait after wake, reclaim, and reblock.
func (r *Repository) RescheduleIdentityConvergenceProbeRetry(ctx context.Context, workID, dependencyKeyHash string, observedProbeAfter, now time.Time) error {
	workID = strings.TrimSpace(workID)
	dependencyKeyHash = strings.TrimSpace(dependencyKeyHash)
	if workID == "" || dependencyKeyHash == "" || observedProbeAfter.IsZero() || now.IsZero() {
		return ErrInvalidWorkItem
	}
	_, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET DependencyProbeAfter = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND DependencyKeyHash = ?
   AND DependencyProbeAfter = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		now.Add(identityConvergenceProbeRetryDelay), now,
		workID, dependencyKeyHash, observedProbeAfter, WorkStateProcessing,
	)
	return err
}

// ScheduleRemoteIdentityConvergenceProbeRetry marks only waiters whose exact
// finalized-link wake delivery failed. Ordinary unresolved dependencies remain
// event-driven and are never enrolled in steady-state polling.
func (r *Repository) ScheduleRemoteIdentityConvergenceProbeRetry(ctx context.Context, signal IdentityConvergenceRemoteSignal, now time.Time) error {
	signal.ConnectorOwnerKey = strings.TrimSpace(signal.ConnectorOwnerKey)
	signal.ConnectorNamespace = strings.TrimSpace(signal.ConnectorNamespace)
	signal.DestinationSlotID = strings.TrimSpace(signal.DestinationSlotID)
	signal.ConnectorRemoteID = strings.TrimSpace(signal.ConnectorRemoteID)
	if now.IsZero() ||
		signal.ConnectorOwnerKey == "" ||
		signal.ConnectorNamespace == "" ||
		signal.DestinationSlotID == "" ||
		signal.ConnectorRemoteID == "" {
		return ErrInvalidWorkItem
	}
	_, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET DependencyProbeAfter = ?,
       LastModifiedDate = ?
 WHERE DependencyKind = ?
   AND DependencyConnectorOwnerKey = ?
   AND DependencyConnectorNamespace = ?
   AND DependencyDestinationSlotId = ?
   AND DependencyConnectorRemoteId = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		now.Add(identityConvergenceProbeRetryDelay), now,
		contractmodels.FlowDependencyKindIdentityConvergence,
		signal.ConnectorOwnerKey,
		signal.ConnectorNamespace,
		signal.DestinationSlotID,
		signal.ConnectorRemoteID,
		WorkStateProcessing,
	)
	return err
}

// ScheduleRecordIdentityConvergenceProbeRetry preserves a one-shot retry when
// the exact local-record wake update itself is temporarily unavailable.
func (r *Repository) ScheduleRecordIdentityConvergenceProbeRetry(ctx context.Context, objectAPIName, recordID string, now time.Time) error {
	objectAPIName = strings.TrimSpace(objectAPIName)
	recordID = strings.TrimSpace(recordID)
	if now.IsZero() || objectAPIName == "" || recordID == "" {
		return ErrInvalidWorkItem
	}
	_, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET DependencyProbeAfter = ?,
       LastModifiedDate = ?
 WHERE DependencyKind = ?
   AND DependencyObjectAPIName = ?
   AND DependencyRecordId = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		now.Add(identityConvergenceProbeRetryDelay), now,
		contractmodels.FlowDependencyKindIdentityConvergence,
		objectAPIName, recordID,
		WorkStateProcessing,
	)
	return err
}

func (r *Repository) scheduleIdentityConvergenceProbe(ctx context.Context, workID, dependencyKeyHash string, probeAfter, now time.Time) error {
	workID = strings.TrimSpace(workID)
	dependencyKeyHash = strings.TrimSpace(dependencyKeyHash)
	if workID == "" || dependencyKeyHash == "" || probeAfter.IsZero() || now.IsZero() || probeAfter.Before(now) {
		return ErrInvalidWorkItem
	}
	_, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET DependencyProbeAfter = ?,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND DependencyKeyHash = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		probeAfter, now,
		workID, dependencyKeyHash, WorkStateProcessing,
	)
	return err
}

// ClearIdentityConvergenceProbeRetry removes an observed fallback deadline.
//
// Deprecated: identity convergence supervisors should keep unresolved waits
// durable with ScheduleIdentityConvergenceProbeRetry. This method remains for
// source compatibility with published workqueue consumers and compares the
// observed deadline so a stale caller cannot clear a newer schedule.
func (r *Repository) ClearIdentityConvergenceProbeRetry(ctx context.Context, workID, dependencyKeyHash string, observedProbeAfter, now time.Time) error {
	workID = strings.TrimSpace(workID)
	dependencyKeyHash = strings.TrimSpace(dependencyKeyHash)
	if workID == "" || dependencyKeyHash == "" || observedProbeAfter.IsZero() || now.IsZero() {
		return ErrInvalidWorkItem
	}
	_, err := r.db.ExecContext(ctx, `
UPDATE _System_ConnectorWorkItem
   SET DependencyProbeAfter = NULL,
       LastModifiedDate = ?
 WHERE WorkId = ?
   AND DependencyKeyHash = ?
   AND DependencyProbeAfter = ?
   AND State = ?
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL`,
		now, workID, dependencyKeyHash, observedProbeAfter, WorkStateProcessing,
	)
	return err
}

// ListRecordDependencies returns a bounded, deduplicated set of persisted
// identity-convergence selectors for startup recovery. Raw
// selectors stay inside the resolver boundary and must not be logged.
func (r *Repository) ListRecordDependencies(ctx context.Context, limit int) ([]contractmodels.FlowExecutionDependency, error) {
	if limit <= 0 || limit > maxDependencyReconcileBatch {
		return nil, fmt.Errorf("connector workqueue: dependency reconciliation limit must be between 1 and %d", maxDependencyReconcileBatch)
	}
	rows, err := r.db.QueryContext(ctx, `
SELECT DependencyKind,
       DependencyObjectAPIName,
       DependencyRecordId,
       DependencyKeyHash,
       DependencyConnectorOwnerKey,
       DependencyConnectorNamespace,
       DependencyConnectorSourceObject,
       DependencyDestinationSlotId,
       DependencyConnectorRemoteId
  FROM _System_ConnectorWorkItem
 WHERE State = ?
   AND DependencyKind IS NOT NULL
   AND LeaseOwner IS NULL
   AND LeaseToken IS NULL
   AND LeaseUntil IS NULL
 ORDER BY DependencyBlockedAt ASC, WorkId ASC
 LIMIT ?`, WorkStateProcessing, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	seen := make(map[string]struct{})
	dependencies := make([]contractmodels.FlowExecutionDependency, 0)
	for rows.Next() {
		var dependency contractmodels.FlowExecutionDependency
		if err := rows.Scan(
			&dependency.Kind,
			&dependency.ObjectAPIName,
			&dependency.RecordID,
			&dependency.KeyHash,
			&dependency.ConnectorOwnerKey,
			&dependency.ConnectorNamespace,
			&dependency.ConnectorSourceObject,
			&dependency.DestinationSlotID,
			&dependency.ConnectorRemoteID,
		); err != nil {
			return nil, err
		}
		if err := dependency.Validate(); err != nil {
			return nil, fmt.Errorf("connector workqueue: invalid persisted dependency: %w", err)
		}
		if _, ok := seen[dependency.KeyHash]; ok {
			continue
		}
		seen[dependency.KeyHash] = struct{}{}
		dependencies = append(dependencies, dependency)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return dependencies, nil
}

// ListIdentityConvergenceWaits supplies restart probes with both
// the exact record candidate and the immutable connector source locator.
func (r *Repository) ListIdentityConvergenceWaits(ctx context.Context, after *IdentityConvergenceCursor, limit int) ([]IdentityConvergenceWait, error) {
	if limit <= 0 || limit > maxDependencyReconcileBatch {
		return nil, fmt.Errorf("connector workqueue: dependency reconciliation limit must be between 1 and %d", maxDependencyReconcileBatch)
	}
	args := []any{WorkStateProcessing}
	cursorPredicate := ""
	if after != nil {
		if after.DependencyBlockedAt.IsZero() || strings.TrimSpace(after.WorkID) == "" {
			return nil, fmt.Errorf("connector workqueue: dependency reconciliation cursor is invalid")
		}
		cursorPredicate = `
   AND (w.DependencyBlockedAt > ? OR (w.DependencyBlockedAt = ? AND w.WorkId > ?))`
		args = append(args, after.DependencyBlockedAt, after.DependencyBlockedAt, strings.TrimSpace(after.WorkID))
	}
	args = append(args, limit)
	rows, err := r.db.QueryContext(ctx, `
SELECT w.WorkId,
       w.InstalledPackId,
       w.DependencyBlockedAt,
       e.Namespace,
       e.SourceObjectKind,
       e.SourceObjectID,
       w.DependencyKind,
       w.DependencyObjectAPIName,
       w.DependencyRecordId,
       w.DependencyKeyHash,
       w.DependencyConnectorOwnerKey,
       w.DependencyConnectorNamespace,
       w.DependencyConnectorSourceObject,
       w.DependencyDestinationSlotId,
       w.DependencyConnectorRemoteId,
       w.DependencyProbeAfter
  FROM _System_ConnectorWorkItem w
  JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.State = ?
   AND w.DependencyKind = `+identityConvergenceDependencyKindSQL+`
   AND w.LeaseOwner IS NULL
   AND w.LeaseToken IS NULL
   AND w.LeaseUntil IS NULL`+cursorPredicate+`
 ORDER BY w.DependencyBlockedAt ASC, w.WorkId ASC
 LIMIT ?`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	return scanIdentityConvergenceWaits(rows)
}

// ListDueIdentityConvergenceWaits returns waits whose persisted fallback probe
// is due. Absolute expiry is owned by ListDueOrExpiredIdentityConvergenceWaits.
func (r *Repository) ListDueIdentityConvergenceWaits(ctx context.Context, now time.Time, limit int) ([]IdentityConvergenceWait, error) {
	return r.listDueIdentityConvergenceWaits(ctx, now, time.Time{}, limit)
}

// ListDueOrExpiredIdentityConvergenceWaits returns waits whose persisted
// fallback probe is due plus expired waits that have never scheduled a probe.
// Once a probe failure persists a future DependencyProbeAfter, that durable
// backoff remains authoritative even after the convergence deadline; expiry
// cannot bypass it and hot-loop an unavailable owner.
func (r *Repository) ListDueOrExpiredIdentityConvergenceWaits(ctx context.Context, now, expireBefore time.Time, limit int) ([]IdentityConvergenceWait, error) {
	if expireBefore.IsZero() {
		return nil, fmt.Errorf("connector workqueue: dependency expiry cutoff is required")
	}
	return r.listDueIdentityConvergenceWaits(ctx, now, expireBefore, limit)
}

func (r *Repository) listDueIdentityConvergenceWaits(ctx context.Context, now, expireBefore time.Time, limit int) ([]IdentityConvergenceWait, error) {
	if now.IsZero() || limit <= 0 || limit > maxDependencyReconcileBatch {
		return nil, fmt.Errorf("connector workqueue: due dependency reconciliation input is invalid")
	}
	expiryPredicate := ""
	orderBy := "w.DependencyProbeAfter ASC, w.WorkId ASC"
	args := []any{WorkStateProcessing, now}
	if !expireBefore.IsZero() {
		expiryPredicate = `
		OR (w.DependencyBlockedAt <= ?
		    AND w.DependencyProbeAfter IS NULL)`
		orderBy = `CASE WHEN w.DependencyBlockedAt <= ? THEN 0 ELSE 1 END ASC,
          CASE WHEN w.DependencyBlockedAt <= ? THEN w.DependencyBlockedAt END ASC,
          w.DependencyProbeAfter ASC, w.WorkId ASC`
		args = append(args, expireBefore)
		args = append(args, expireBefore, expireBefore)
	}
	args = append(args, limit)
	rows, err := r.db.QueryContext(ctx, `
SELECT w.WorkId,
       w.InstalledPackId,
       w.DependencyBlockedAt,
       COALESCE(e.Namespace, ''),
       COALESCE(e.SourceObjectKind, ''),
       COALESCE(e.SourceObjectID, ''),
       w.DependencyKind,
       w.DependencyObjectAPIName,
       w.DependencyRecordId,
       w.DependencyKeyHash,
       w.DependencyConnectorOwnerKey,
       w.DependencyConnectorNamespace,
       w.DependencyConnectorSourceObject,
       w.DependencyDestinationSlotId,
       w.DependencyConnectorRemoteId,
       w.DependencyProbeAfter
  FROM _System_ConnectorWorkItem w
  LEFT JOIN _System_ConnectorEvent e
    ON e.Id = w.EventId
   AND e.InstalledPackId = w.InstalledPackId
 WHERE w.State = ?
   AND w.DependencyKind = `+identityConvergenceDependencyKindSQL+`
   AND w.LeaseOwner IS NULL
   AND w.LeaseToken IS NULL
   AND w.LeaseUntil IS NULL
   AND ((w.DependencyProbeAfter IS NOT NULL
         AND w.DependencyProbeAfter <= ?)`+expiryPredicate+`)
 ORDER BY `+orderBy+`
 LIMIT ?`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanIdentityConvergenceWaits(rows)
}

func scanIdentityConvergenceWaits(rows *sql.Rows) ([]IdentityConvergenceWait, error) {
	waits := make([]IdentityConvergenceWait, 0)
	for rows.Next() {
		var wait IdentityConvergenceWait
		if err := rows.Scan(
			&wait.WorkID,
			&wait.InstalledPackID,
			&wait.DependencyBlockedAt,
			&wait.Namespace,
			&wait.ObjectKind,
			&wait.SourceID,
			&wait.Dependency.Kind,
			&wait.Dependency.ObjectAPIName,
			&wait.Dependency.RecordID,
			&wait.Dependency.KeyHash,
			&wait.Dependency.ConnectorOwnerKey,
			&wait.Dependency.ConnectorNamespace,
			&wait.Dependency.ConnectorSourceObject,
			&wait.Dependency.DestinationSlotID,
			&wait.Dependency.ConnectorRemoteID,
			&wait.DependencyProbeAfter,
		); err != nil {
			return nil, err
		}
		if IsIdentityConvergenceWaitInvalid(wait) {
			log.Printf(
				"WARN connector workqueue: skipped invalid persisted identity convergence wait work_id=%s failure_class=%s",
				strings.TrimSpace(wait.WorkID),
				FailureClassIdentityConvergenceInvalid,
			)
			waits = append(waits, wait)
			continue
		}
		waits = append(waits, wait)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return waits, nil
}