package connector_outboundintent

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"sort"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

const maxDependencyReconcileBatch = 1000

// The indexable equality bounds the read; the binary residual preserves exact
// identity even when the physical LinkKey collation is case-insensitive.
const remoteLinkDependencyLookupSQL = `SELECT 1
  FROM _System_ConnectorRemoteLink
 WHERE LinkKey = ? AND BINARY LinkKey = BINARY ?
 LIMIT 1`

func associationRemoteLinkDependency(
	scope Scope,
	connectorOwnerKey string,
	sourceObject string,
	sourceRecordID string,
	destinationSlotID string,
	readinessKeys []string,
) commandDependency {
	logicalScope := integrationLogicalEffectScope(scope)
	return commandDependency{
		Kind: dependencyKindAssociationRemoteLink,
		Key: "integration_association_dependency_v1:" + digestStrings(
			logicalScope.OrgID, logicalScope.AppInstallationID,
			logicalScope.AuthorizedWorkloadID, connectorOwnerKey,
			connectorPushActionType, sourceObject, sourceRecordID, destinationSlotID,
		),
		ReadinessKeys: append([]string(nil), readinessKeys...),
	}
}

func remoteLinkDependencyReady(
	ctx context.Context,
	tx *sql.Tx,
	readinessKeys []string,
) (bool, error) {
	keys := append([]string(nil), readinessKeys...)
	sort.Strings(keys)
	for _, key := range keys {
		var present int
		err := tx.QueryRowContext(ctx, remoteLinkDependencyLookupSQL+" FOR UPDATE", key, key).Scan(&present)
		switch {
		case err == nil:
			return true, nil
		case errors.Is(err, sql.ErrNoRows):
		default:
			return false, err
		}
	}
	return false, nil
}

// waitForDependency removes a pre-dispatch command from the runnable queue.
// An absent Remote Link is not a TiDB gap lock, so the first transaction uses
// a runnable registration state. A post-commit reconciliation settles it to a
// wait or retry; a crash leaves ordinary scheduler recovery, never a lost wake.
func (r *Repository) waitForDependency(
	ctx context.Context,
	claim Claim,
	dependency commandDependency,
	code string,
) (Command, error) {
	if r == nil || r.db == nil || ctx == nil || validateClaim(claim) != nil ||
		validateDependency(dependency) != nil || dependency.Kind != dependencyKindAssociationRemoteLink ||
		!safeCode(code) {
		return Command{}, ErrInvalidInput
	}
	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())
	}

	dependencyReady, err := remoteLinkDependencyReady(ctx, tx, dependency.ReadinessKeys)
	if err != nil {
		return rollback(err)
	}

	locked, err := scanCommand(tx.QueryRowContext(
		ctx, commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`, claim.Command.CommandID,
	))
	if err != nil {
		return rollback(err)
	}
	now := r.now().UTC()
	readinessKeys, err := json.Marshal(dependency.ReadinessKeys)
	if err != nil {
		return rollback(err)
	}
	if locked.State != StateClaimed || locked.ClaimToken != claim.Token ||
		locked.LeaseExpiresAt == nil || !locked.LeaseExpiresAt.After(now) ||
		locked.DispatchStarted != nil || !sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}

	nextState := stateDependencyRegistration
	var nextAttempt, dependencyKind, dependencyKey, persistedReadinessKeys, blockedAt any
	if dependencyReady {
		nextState = StateRetry
		nextAttempt = now
	} else {
		nextAttempt = now
		dependencyKind = dependency.Kind
		dependencyKey = dependency.Key
		persistedReadinessKeys = readinessKeys
		blockedAt = now
	}
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, DispatchStartedAt = NULL, NextAttemptAt = ?,
       DependencyKind = ?, DependencyKey = ?, DependencyReadinessKeys = ?, DependencyBlockedAt = ?,
       FailureCode = CASE
           WHEN RemoteId IS NOT NULL AND FailureCode IN (?, ?) THEN FailureCode
           ELSE ?
       END,
       LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ? AND DispatchStartedAt IS NULL`,
		nextState, nextAttempt, dependencyKind, dependencyKey, persistedReadinessKeys, blockedAt,
		"association_create_outcome_uncertain", "association_update_outcome_uncertain", code,
		now, locked.CommandID, StateClaimed, 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
	}
	if !dependencyReady {
		if err := r.settleRemoteLinkDependencyRegistration(
			ctx, locked.CommandID, locked.Scope, dependency, r.now().UTC(),
		); err != nil {
			return Command{}, err
		}
	}
	return r.loadByID(ctx, locked.CommandID)
}

func (r *Repository) settleRemoteLinkDependencyRegistration(
	ctx context.Context,
	commandID string,
	scope Scope,
	dependency commandDependency,
	now time.Time,
) error {
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) error { return errors.Join(cause, tx.Rollback()) }

	dependencyReady, err := remoteLinkDependencyReady(ctx, tx, dependency.ReadinessKeys)
	if err != nil {
		return rollback(err)
	}
	var state, dependencyKind, dependencyKey string
	err = tx.QueryRowContext(ctx, `
SELECT State, COALESCE(DependencyKind, ''), COALESCE(DependencyKey, '')
  FROM _System_IntegrationOutboundCommand
 WHERE Id = ? AND ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ?
   AND AuthorizedWorkloadId = ?
 LIMIT 1 FOR UPDATE`,
		commandID, scopeDigest(scope), scope.OrgID, scope.AppInstallationID,
		scope.AuthorizedWorkloadID,
	).Scan(&state, &dependencyKind, &dependencyKey)
	if err != nil {
		return rollback(err)
	}
	if state != stateDependencyRegistration || dependencyKind != dependency.Kind ||
		dependencyKey != dependency.Key {
		if err := tx.Commit(); err != nil {
			return err
		}
		return nil
	}

	nextState := stateWaitingDependency
	var nextAttempt any
	clearDependency := false
	if dependencyReady {
		nextState = StateRetry
		nextAttempt = now
		clearDependency = true
	}
	_, err = tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, NextAttemptAt = ?,
       DependencyKind = CASE WHEN ? THEN NULL ELSE DependencyKind END,
       DependencyKey = CASE WHEN ? THEN NULL ELSE DependencyKey END,
       DependencyReadinessKeys = CASE WHEN ? THEN NULL ELSE DependencyReadinessKeys END,
       DependencyBlockedAt = CASE WHEN ? THEN NULL ELSE DependencyBlockedAt END,
       LastModifiedDate = ?
 WHERE Id = ? AND ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ?
   AND AuthorizedWorkloadId = ? AND State = ? AND DependencyKind = ?
   AND DependencyKey = ?`,
		nextState, nextAttempt, clearDependency, clearDependency, clearDependency, clearDependency, now,
		commandID, scopeDigest(scope), scope.OrgID, scope.AppInstallationID,
		scope.AuthorizedWorkloadID, stateDependencyRegistration,
		dependency.Kind, dependency.Key,
	)
	if err != nil {
		return rollback(err)
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	return nil
}

func wakeRemoteLinkDependents(
	ctx context.Context,
	tx *sql.Tx,
	finalization RemoteLinkFinalization,
	now time.Time,
) error {
	dependency := associationRemoteLinkDependency(
		finalization.Scope, finalization.ConnectorOwnerKey, finalization.SourceObject,
		finalization.SourceRecordID, finalization.DestinationSlotID,
		[]string{finalization.LinkKey},
	)
	if ctx == nil || tx == nil || validateDependency(dependency) != nil {
		return ErrInvalidInput
	}
	_, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, NextAttemptAt = ?, DependencyKind = NULL,
       DependencyKey = NULL, DependencyReadinessKeys = NULL,
       DependencyBlockedAt = NULL, LastModifiedDate = ?
 WHERE ScopeDigest = ? AND OrgId = ? AND AppInstallationId = ?
   AND AuthorizedWorkloadId = ? AND State IN (?, ?) AND DependencyKind = ?
   AND DependencyKey = ?`,
		StateRetry, now, now, scopeDigest(finalization.Scope), finalization.Scope.OrgID,
		finalization.Scope.AppInstallationID, finalization.Scope.AuthorizedWorkloadID,
		stateDependencyRegistration, stateWaitingDependency,
		dependencyKindAssociationRemoteLink, dependency.Key,
	)
	return err
}

type dependencyWait struct {
	commandID     string
	dependencyKey string
	readinessKeys []string
	blockedAt     time.Time
}

const dependencyReconcileListSQL = `
SELECT Id, DependencyKey, DependencyReadinessKeys, DependencyBlockedAt
  FROM _System_IntegrationOutboundCommand USE INDEX (idx_integration_outbound_dependency_reconcile)
 WHERE ScopeDigest = ? AND CatalogRevision = ? AND State = ?
   AND DependencyKind = ? AND Id > ?
 ORDER BY Id
 LIMIT ?`

// reconcileDependencyWaits is a low-frequency safety net for a producer wake
// that never reached a durable waiter. It reads only the exact wait index and
// exact Remote Link keys; it never reopens command preparation or retry state.
func (r *Repository) reconcileDependencyWaits(
	ctx context.Context,
	scope Scope,
	afterID string,
	limit int,
) (string, bool, int, int, error) {
	if r == nil || r.db == nil || ctx == nil || scope.validate() != nil ||
		limit <= 0 || limit > maxDependencyReconcileBatch || afterID != strings.TrimSpace(afterID) {
		return "", false, 0, 0, ErrInvalidInput
	}
	inspected := 0
	rows, err := r.db.QueryContext(ctx, dependencyReconcileListSQL,
		scopeDigest(scope), scope.CatalogRevision, stateWaitingDependency,
		dependencyKindAssociationRemoteLink, afterID, limit+1,
	)
	if err != nil {
		return afterID, true, 0, inspected, err
	}
	defer rows.Close()
	waits := make([]dependencyWait, 0, limit+1)
	for rows.Next() {
		var wait dependencyWait
		var encoded []byte
		if err := rows.Scan(&wait.commandID, &wait.dependencyKey, &encoded, &wait.blockedAt); err != nil {
			return afterID, true, 0, inspected, err
		}
		if err := json.Unmarshal(encoded, &wait.readinessKeys); err != nil ||
			validateDependency(commandDependency{
				Kind: dependencyKindAssociationRemoteLink, Key: wait.dependencyKey,
				ReadinessKeys: wait.readinessKeys,
			}) != nil || wait.blockedAt.IsZero() {
			return afterID, true, 0, inspected, ErrSchemaUnverified
		}
		waits = append(waits, wait)
	}
	if err := rows.Err(); err != nil {
		return afterID, true, 0, inspected, err
	}
	more := len(waits) > limit
	if more {
		waits = waits[:limit]
	}
	woken := 0
	for _, wait := range waits {
		inspected++
		ready, err := remoteLinkDependencyPresent(ctx, r.db, wait.readinessKeys)
		if err != nil {
			return afterID, true, woken, inspected, err
		}
		if !ready {
			afterID = wait.commandID
			continue
		}
		now := r.now().UTC()
		result, err := r.db.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, NextAttemptAt = ?, DependencyKind = NULL,
       DependencyKey = NULL, DependencyReadinessKeys = NULL,
       DependencyBlockedAt = NULL, LastModifiedDate = ?
 WHERE Id = ? AND ScopeDigest = ? AND CatalogRevision = ? AND State = ?
   AND DependencyKind = ? AND DependencyKey = ? AND DependencyBlockedAt = ?`,
			StateRetry, now, now, wait.commandID,
			scopeDigest(scope), scope.CatalogRevision, stateWaitingDependency,
			dependencyKindAssociationRemoteLink, wait.dependencyKey, wait.blockedAt,
		)
		if err != nil {
			return afterID, true, woken, inspected, err
		}
		changed, err := result.RowsAffected()
		if err != nil {
			return afterID, true, woken, inspected, err
		}
		if changed == 1 {
			woken++
		} else if changed != 0 {
			return afterID, true, woken, inspected, ErrSchemaUnverified
		}
		afterID = wait.commandID
	}
	if len(waits) == 0 || !more {
		return "", false, woken, inspected, nil
	}
	return waits[len(waits)-1].commandID, true, woken, inspected, nil
}

func remoteLinkDependencyPresent(ctx context.Context, db *sql.DB, readinessKeys []string) (bool, error) {
	keys := append([]string(nil), readinessKeys...)
	sort.Strings(keys)
	for _, key := range keys {
		var present int
		err := db.QueryRowContext(ctx, remoteLinkDependencyLookupSQL, key, key).Scan(&present)
		switch {
		case err == nil:
			return true, nil
		case errors.Is(err, sql.ErrNoRows):
		default:
			return false, err
		}
	}
	return false, nil
}