package connector_events

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

	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

// ClaimNextCoalescedSibling atomically claims the immediately-following row in
// the same source lane and then marks the currently-held same-source row
// processed.
// It is intentionally narrow: callers must already hold currentEventID's
// processing lease, and only the next non-terminal sibling with the same
// InstalledPackId, EventKind, RouteVersion, companion identity, objectKind, and
// sourceID can be claimed.
func (s *Service) ClaimNextCoalescedSibling(ctx context.Context, currentEventID, currentClaimToken, workerID string) (*ConnectorEvent, bool, error) {
	return s.claimNextCoalescedSibling(ctx, currentEventID, currentClaimToken, workerID, false)
}

// ClaimNextCoalescedSiblingSourceScoped is the versioned handoff used only by
// routes that explicitly opt into cross-source delivery independence.
func (s *Service) ClaimNextCoalescedSiblingSourceScoped(ctx context.Context, currentEventID, currentClaimToken, workerID string) (*ConnectorEvent, bool, error) {
	return s.claimNextCoalescedSibling(ctx, currentEventID, currentClaimToken, workerID, true)
}

func (s *Service) claimNextCoalescedSibling(ctx context.Context, currentEventID, currentClaimToken, workerID string, allowCrossSourceDelivery bool) (*ConnectorEvent, bool, error) {
	currentEventID = strings.TrimSpace(currentEventID)
	currentClaimToken = strings.TrimSpace(currentClaimToken)
	workerID = strings.TrimSpace(workerID)
	if currentEventID == "" || currentClaimToken == "" || workerID == "" {
		return nil, false, fmt.Errorf("connector events: coalesce claim requires current event, claim token, and worker")
	}

	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return nil, false, err
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
		_ = cleanup()
	}()

	now := s.now()
	var current coalesceCursor
	currentQuery := `
SELECT InstalledPackId, EventKind, COALESCE(RouteVersion, 0),
       COALESCE(CompanionPackInstallId, ''), COALESCE(CompanionFlowVersionId, ''),
       COALESCE(SourceLaneKey, ''),
       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.objectKind')), ''),
       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(Payload, '$.sourceID')), ''),
       DeliverySequence, EnqueueOrdinal, CreatedDate
  FROM _System_ConnectorEvent e
 WHERE e.Id = ?
   AND e.Status = ?
   AND e.ClaimToken = ?`
	currentGuardMode := legacyEventWorkItemGuardFull
	for {
		guardSQL, guardArgs := s.legacyEventWorkItemGuardSQL("e", currentGuardMode)
		args := append([]any{currentEventID, statusProcessing, currentClaimToken}, guardArgs...)
		err = tx.QueryRowContext(ctx, currentQuery+guardSQL+` FOR UPDATE`, args...).Scan(
			&current.installedPackID,
			&current.eventKind,
			&current.routeVersion,
			&current.companionPackInstallID,
			&current.companionFlowVersionID,
			&current.sourceLaneKey,
			&current.objectKind,
			&current.sourceID,
			&current.deliverySequence,
			&current.enqueueOrdinal,
			&current.createdDate,
		)
		nextMode, retry := nextLegacyEventWorkItemGuardMode(currentGuardMode, err)
		if retry {
			currentGuardMode = nextMode
			continue
		}
		break
	}
	if errors.Is(err, sql.ErrNoRows) {
		return nil, false, ErrLeaseLost
	}
	if err != nil {
		return nil, false, err
	}
	if current.objectKind == "" || current.sourceID == "" {
		if err := tx.Commit(); err != nil {
			return nil, false, err
		}
		committed = true
		return nil, false, nil
	}

	next, ok, err := s.nextCoalescibleSibling(ctx, tx, currentEventID, current, allowCrossSourceDelivery)
	if err != nil || !ok {
		if err != nil {
			return nil, false, err
		}
		if err := tx.Commit(); err != nil {
			return nil, false, err
		}
		committed = true
		return nil, false, nil
	}

	freshToken := s.idgen()
	guardSQL, guardBinds := buildClaimOrderingGuardExceptCurrentSQL(currentEventID, now, allowCrossSourceDelivery)
	updateArgs := append([]interface{}{statusProcessing, workerID, now, freshToken, now,
		next.id, statusPending, now, statusFailed, now}, guardBinds...)
	res, err := s.execLegacyEventMutation(ctx, tx, `
	UPDATE _System_ConnectorEvent e
	   SET Status = ?,
	       ClaimedBy = ?,
	       ClaimedAt = ?,
	       ClaimToken = ?,
	       LastModifiedDate = ?
	 WHERE e.Id = ?
	   AND (
	     (e.Status = ? AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
	     OR (e.Status = ? AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
	   )
	   AND `+guardSQL,
		updateArgs...)
	if err != nil {
		return nil, false, err
	}
	if n, err := res.RowsAffected(); err != nil {
		return nil, false, err
	} else if n == 0 {
		if err := tx.Commit(); err != nil {
			return nil, false, err
		}
		committed = true
		return nil, false, nil
	}

	res, err = s.execLegacyEventMutation(ctx, tx, `
	UPDATE _System_ConnectorEvent e
	   SET Status = ?,
	       ProcessedAt = ?,
       LastModifiedDate = ?
 WHERE e.Id = ?
   AND e.Status = ?
   AND e.ClaimToken = ?`,
		statusProcessed, now, now, currentEventID, statusProcessing, currentClaimToken)
	if err != nil {
		return nil, false, err
	}
	if n, err := res.RowsAffected(); err != nil {
		return nil, false, err
	} else if n == 0 {
		return nil, false, ErrLeaseLost
	}

	ev, err := loadByIDTx(ctx, tx, next.id)
	if err != nil {
		return nil, false, err
	}
	if err := tx.Commit(); err != nil {
		return nil, false, err
	}
	committed = true
	ev.ClaimToken = freshToken
	return ev, true, nil
}

type coalesceCursor struct {
	id                     string
	installedPackID        string
	eventKind              string
	routeVersion           int64
	companionPackInstallID string
	companionFlowVersionID string
	sourceLaneKey          string
	objectKind             string
	sourceID               string
	status                 string
	deliverySequence       int64
	enqueueOrdinal         int64
	createdDate            time.Time
}

func (s *Service) nextCoalescibleSibling(ctx context.Context, tx *sql.Tx, currentEventID string, current coalesceCursor, allowCrossSourceDelivery bool) (coalesceCursor, bool, error) {
	return s.nextCoalescibleSiblingWithEventGuard(
		ctx, tx, currentEventID, current, allowCrossSourceDelivery,
		legacyEventWorkItemGuardFull, -1,
	)
}

// nextCoalescibleWorkItemSibling reads the immutable Event payload for the
// next lane member without treating mutable Event executor fields as claim
// authority. The caller must win the corresponding WorkItem CAS in the same
// transaction before treating the Event as executable.
func (s *Service) nextCoalescibleWorkItemSibling(ctx context.Context, tx *sql.Tx, currentEventID string, current coalesceCursor, allowCrossSourceDelivery bool, generation int64) (coalesceCursor, bool, error) {
	return s.nextCoalescibleSiblingWithEventGuard(
		ctx, tx, currentEventID, current, allowCrossSourceDelivery,
		legacyEventWorkItemGuardNone, generation,
	)
}

func (s *Service) nextCoalescibleSiblingWithEventGuard(
	ctx context.Context,
	tx *sql.Tx,
	currentEventID string,
	current coalesceCursor,
	allowCrossSourceDelivery bool,
	initialGuardMode legacyEventWorkItemGuardMode,
	workItemGeneration int64,
) (coalesceCursor, bool, error) {
	var next coalesceCursor
	coalesceSourceLaneKey := coalescibleSourceLaneKey(current)
	now := s.now()
	candidatePredicateSQL := `(
	     (e.Status = ? AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
	     OR (e.Status = ? AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
	   )`
	candidateArgs := []any{statusPending, now, statusFailed, now}
	if initialGuardMode == legacyEventWorkItemGuardNone {
		// The WorkItem coalescer accepts either a fresh Event that can win the
		// one-time ownership transfer or a frozen Event whose matching generation
		// has no WorkItem row yet (a fail-closed coverage gap) or still has a
		// nonterminal executor WorkItem. Frozen Events backed only by terminal
		// WorkItems are immutable history and must not hide a newer successor.
		// The WorkItem repository rechecks the full claim fence in the epoch CAS.
		candidatePredicateSQL = `(
	     (
	       e.ExecutionOwner = ?
	       AND (
	         (e.Status = ? AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
	         OR (e.Status = ? AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
	       )
	     )
	     OR (
	       e.ExecutionOwner = ?
	       AND e.Status = ?
	       AND (
	         NOT EXISTS (
	           SELECT 1
	             FROM _System_ConnectorWorkItem frozen_work
	            WHERE frozen_work.EventId = e.Id
	              AND frozen_work.InstalledPackId = e.InstalledPackId
	              AND frozen_work.Generation = ?
	              AND frozen_work.WorkType IN (?, ?)
	         )
	         OR EXISTS (
	           SELECT 1
	             FROM _System_ConnectorWorkItem frozen_live_work
	            WHERE frozen_live_work.EventId = e.Id
	              AND frozen_live_work.InstalledPackId = e.InstalledPackId
	              AND frozen_live_work.Generation = ?
	              AND frozen_live_work.WorkType IN (?, ?)
	              AND frozen_live_work.State NOT IN (?, ?, ?)
	         )
	       )
	     )
	   )`
		candidateArgs = []any{
			connector_workqueue.EventExecutionOwnerEvent,
			statusPending, now,
			statusFailed, now,
			connector_workqueue.EventExecutionOwnerWorkItem,
			connector_workqueue.EventWorkItemCompatibilityFenceStatus,
			workItemGeneration,
			connector_workqueue.WorkTypeConnectorEvent,
			connector_workqueue.WorkTypeWorkflowActionApply,
			workItemGeneration,
			connector_workqueue.WorkTypeConnectorEvent,
			connector_workqueue.WorkTypeWorkflowActionApply,
			connector_workqueue.WorkStateProcessed,
			connector_workqueue.WorkStateSkipped,
			connector_workqueue.WorkStateDead,
		}
	}
	// NCRM #3139 (Problem A): the ExceptCurrent guard's source-lane arms now carry
	// the NextAttemptAt-gated blocker (bound to `now`), so its binds are spliced
	// at the guard's textual position between the pre-guard predicate args and the
	// trailing ORDER-BY args.
	guardSQL, guardBinds := buildClaimOrderingGuardExceptCurrentSQL(currentEventID, now, allowCrossSourceDelivery)
	args := []interface{}{
		current.installedPackID,
		coalesceSourceLaneKey,
		coalesceSourceLaneKey,
		coalesceSourceLaneKey,
		current.objectKind,
		current.sourceID,
	}
	args = append(args, candidateArgs...)
	args = append(args,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
		currentEventID,
		current.enqueueOrdinal,
		current.createdDate,
		current.createdDate,
		current.deliverySequence,
		current.createdDate,
		current.deliverySequence,
		currentEventID,
	)
	args = append(args, guardBinds...)
	args = append(args,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
		current.enqueueOrdinal,
	)
	baseQuery := `
SELECT e.Id, e.InstalledPackId, e.EventKind, COALESCE(e.RouteVersion, 0),
       COALESCE(e.CompanionPackInstallId, ''), COALESCE(e.CompanionFlowVersionId, ''),
       COALESCE(e.SourceLaneKey, ''),
       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.objectKind')), ''),
       COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.sourceID')), ''),
       e.Status, e.DeliverySequence, e.EnqueueOrdinal, e.CreatedDate
  FROM _System_ConnectorEvent e
 WHERE e.InstalledPackId = ?
   AND (
     (NULLIF(?, '') IS NOT NULL AND e.SourceLaneKey = ?)
     OR (
       (NULLIF(?, '') IS NULL OR NULLIF(e.SourceLaneKey, '') IS NULL)
       AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.objectKind')), '') = ?
       AND COALESCE(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.sourceID')), '') = ?
     )
   )
	   AND ` + candidatePredicateSQL + `
   AND (
     NULLIF(e.CompanionPackInstallId, '') IS NULL
     OR EXISTS (
       SELECT 1
         FROM _System_InstalledPack ep
        WHERE ep.PackInstallId = e.CompanionPackInstallId
          AND ep.Status = 'active'
          AND COALESCE(ep.IsDeleted, 0) = 0
     )
   )
   AND (
     (
       e.EnqueueOrdinal <> 0
       AND ? <> 0
       AND (e.EnqueueOrdinal > ? OR (e.EnqueueOrdinal = ? AND e.Id > ?))
     )
     OR (
       (e.EnqueueOrdinal = 0 OR ? = 0)
       AND (
         e.CreatedDate > ?
         OR (e.CreatedDate = ? AND e.DeliverySequence > ?)
         OR (e.CreatedDate = ? AND e.DeliverySequence = ? AND e.Id > ?)
       )
     )
   )
   AND ` + guardSQL + `
 ORDER BY
   CASE WHEN e.EnqueueOrdinal = 0 OR ? = 0 THEN 0 ELSE 1 END ASC,
   CASE WHEN e.EnqueueOrdinal = 0 OR ? = 0 THEN e.CreatedDate END ASC,
   CASE WHEN e.EnqueueOrdinal = 0 OR ? = 0 THEN e.DeliverySequence END ASC,
   CASE WHEN e.EnqueueOrdinal = 0 OR ? = 0 THEN e.Id END ASC,
   CASE WHEN e.EnqueueOrdinal <> 0 AND ? <> 0 THEN e.EnqueueOrdinal END ASC,
   CASE WHEN e.EnqueueOrdinal <> 0 AND ? <> 0 THEN e.Id END ASC
 LIMIT 1
 FOR UPDATE`
	mode := initialGuardMode
	for {
		workItemGuardSQL, workItemGuardBinds := s.legacyEventWorkItemGuardSQL("e", mode)
		queryArgs := append(append([]any(nil), args...), workItemGuardBinds...)
		err := tx.QueryRowContext(ctx, strings.Replace(baseQuery, "\n ORDER BY", workItemGuardSQL+"\n ORDER BY", 1), queryArgs...).Scan(
			&next.id,
			&next.installedPackID,
			&next.eventKind,
			&next.routeVersion,
			&next.companionPackInstallID,
			&next.companionFlowVersionID,
			&next.sourceLaneKey,
			&next.objectKind,
			&next.sourceID,
			&next.status,
			&next.deliverySequence,
			&next.enqueueOrdinal,
			&next.createdDate,
		)
		nextMode, retry := nextLegacyEventWorkItemGuardMode(mode, err)
		if retry {
			mode = nextMode
			continue
		}
		if errors.Is(err, sql.ErrNoRows) {
			return coalesceCursor{}, false, nil
		}
		if err != nil {
			return coalesceCursor{}, false, err
		}
		break
	}
	if next.eventKind != current.eventKind ||
		next.routeVersion != current.routeVersion ||
		next.companionPackInstallID != current.companionPackInstallID ||
		next.companionFlowVersionID != current.companionFlowVersionID {
		return coalesceCursor{}, false, nil
	}
	switch next.status {
	case statusPending, statusFailed:
		return next, true, nil
	case connector_workqueue.EventWorkItemCompatibilityFenceStatus:
		if initialGuardMode == legacyEventWorkItemGuardNone {
			return next, true, nil
		}
		return coalesceCursor{}, false, nil
	default:
		return coalesceCursor{}, false, nil
	}
}

func coalescibleSourceLaneKey(current coalesceCursor) string {
	sourceLaneKey := strings.TrimSpace(current.sourceLaneKey)
	if sourceLaneKey == "" {
		return ""
	}
	if sourceLaneKey != sourceLaneKeyFor(current.objectKind, current.sourceID) {
		return ""
	}
	return sourceLaneKey
}

func loadByIDTx(ctx context.Context, tx *sql.Tx, eventID string) (*ConnectorEvent, error) {
	rows, err := tx.QueryContext(ctx, `
SELECT Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status,
       QueueClass, Payload, RetryCount, LastError, LastErrorAt,
       ClaimedBy, ClaimedAt, ClaimToken, ProcessedAt, ReplayedAt, ReplayOfId,
       RouteVersion, CompanionPackInstallId, CompanionFlowVersionId,
       SourceLaneKey, DeliveryId, DeliverySequence, CreatedDate, LastModifiedDate
  FROM _System_ConnectorEvent
 WHERE Id = ?`, eventID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	events, err := scanConnectorEvents(rows)
	if err != nil {
		return nil, err
	}
	if len(events) == 0 {
		return nil, ErrEventNotFound
	}
	return &events[0], nil
}