package connector_events

import "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events/claimpolicy"

// Claim-ordering guard builders.
//
// The guard blocks a candidate row `e` when an earlier non-terminal sibling `b`
// still exists, either:
//  1. within the same DeliveryId by lower DeliverySequence (DELIVERY branch), or
//  2. within the same queue-envelope (objectKind, sourceID) lane by lower
//     pack-scoped EnqueueOrdinal, with Id as a deterministic tiebreak when two
//     rows carry the same ordinal (SOURCE-LANE branch). During mixed-version
//     upgrades, old writers can still create EnqueueOrdinal=0 rows; those rows
//     fall back to the historical queue order (CreatedDate, DeliverySequence, Id)
//     so they cannot skip ahead of already-ordered siblings.
//
// The second branch is intentionally generic: connector_inbound queue rows
// carry objectKind + sourceID at the envelope boundary, and enqueue materializes
// a fixed-width SourceLaneKey so claim can use a normal index instead of
// evaluating JSON_EXTRACT across the fully-upgraded backlog. Some workloads
// dispatch logically-related mutations for one source record as separate
// deliveries. Delivery ordering alone cannot serialize those rows, so enqueue
// allocates a monotonic EnqueueOrdinal while the active pack row is locked. If
// two rows still end up with the same ordinal (for example due to manual
// repair, legacy data, or an allocator race), Id remains a stable single-
// winner tiebreak so only one same-source row is claimable at a time.
//
// The third branch is a mixed-version safety net. During rolling deploys, an old
// writer can enqueue payload-only rows after the one-shot warm-upgrade backfill
// stamped complete. New workers must still serialize those rows with
// materialized rows, so the fallback compares Payload only when either side is
// missing SourceLaneKey. Once all writers are upgraded and the backlog has been
// backfilled, the indexed SourceLaneKey branch is the hot path.
//
// Rows with missing/blank objectKind or sourceID do not enter the same-source
// branch because NULL/empty values fail the explicit NULLIF(..., "") checks.
//
// NextAttemptAt-gated SOURCE-LANE blocker (NCRM #3139, Problem A)
// --------------------------------------------------------------
// The blocker `b` is "non-terminal" by Status alone today. A `failed` row that
// is backing off between retries (NextAttemptAt in the future) is non-terminal
// but NOT itself claimable — the outer candidate arm already gates `failed` on
// `NextAttemptAt <= now`. With a Status-only blocker, such a parked-failed head
// blocked every newer SOURCE-LANE sibling for the whole retry ladder
// (5s/30s/2m/5m, ≈455s to dead), violating per-source no-idle even though the
// pack-wide single-flight invariant held.
//
// The SOURCE-LANE blocker (branches 2 + 3, the cold residual, and the lane-head
// derived table) therefore reads the SAME actionability fact the outer claim arm
// reads: a `failed` blocker stops blocking its lane once its NextAttemptAt is in
// the future. A `failed` blocker whose NextAttemptAt is NULL or due (<= now)
// STILL blocks. The DELIVERY branch (branch 1 + the delivery-head derived table)
// stays STRICT FIFO: a failed delivery-predecessor always blocks its tail
// regardless of NextAttemptAt (link-after-create depends on it).
//
// Because the gated blocker introduces a positional `?` (bound to `now`) into a
// predicate that was embedded ~20 times across 5 files, the guard is built by Go
// functions returning (sql, binds) so each call site splices the SQL and its
// binds together — injecting a bare `?` into a shared const would silently
// misalign positional args.

// blockerTerminalStatus is the strict DELIVERY-branch blocker: a sibling blocks
// while it is in any non-terminal Status. Carries no bind. Used by the delivery
// NOT EXISTS branch (the `b`-aliased subquery).
const blockerTerminalStatus = `b.Status NOT IN ('processed', 'skipped', 'dead')`

// deliveryHeadTerminalStatus is the unaliased form of blockerTerminalStatus for
// the delivery-head derived table, which scans the bare table (no `b.` alias).
const deliveryHeadTerminalStatus = `Status NOT IN ('processed', 'skipped', 'dead')`

// laneBlockerStatusGated is the NextAttemptAt-gated SOURCE-LANE blocker: a
// sibling blocks while it is non-terminal AND (if it is `pending` or `failed`)
// its not-before time is due. A settled pending row or backing-off failed row
// with a future NextAttemptAt no longer blocks its lane. Contributes exactly
// one `?` bound to `now`.
const laneBlockerStatusGated = `(
          b.Status NOT IN ('processed', 'skipped', 'dead', 'pending', 'failed')
          OR (b.Status IN ('pending', 'failed') AND (b.NextAttemptAt IS NULL OR b.NextAttemptAt <= ?))
        )`

// laneHeadBlockerStatusGated is laneBlockerStatusGated re-aliased for the
// lane-head ROW_NUMBER source set, which scans the bare table (no `b.` alias).
const laneHeadBlockerStatusGated = `(
          Status NOT IN ('processed', 'skipped', 'dead', 'pending', 'failed')
          OR (Status IN ('pending', 'failed') AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?))
        )`

// buildClaimOrderingGuard returns the full three-branch ordering guard predicate
// and its binds. Branch 1 (delivery) is strict FIFO; branches 2 (source-lane
// hot) and 3 (JSON fallback) are NextAttemptAt-gated, each contributing one
// `now` bind in textual order: [branch2 now, branch3 now].
func buildClaimOrderingGuard(now interface{}) (string, []interface{}) {
	return claimpolicy.BuildEventOrderingGuard(now)
}

// jsonResidualPredicate is branch 3 of the guard: the JSON-payload same-source
// fallback for rows missing a materialized SourceLaneKey. It carries the gated
// SOURCE-LANE blocker (one `?` for `now`) — same actionability fact as branch 2.
// It only matches when SourceLaneKey is NULL on either side, so it is empty once
// enqueue materializes SourceLaneKey for the whole backlog.
const jsonResidualPredicate = `NOT EXISTS (
     SELECT 1 FROM _System_ConnectorEvent b
      WHERE b.InstalledPackId = e.InstalledPackId
        AND ` + laneBlockerStatusGated + `
        AND b.Id <> e.Id
        AND (
          NULLIF(e.SourceLaneKey, '') IS NULL
          OR NULLIF(b.SourceLaneKey, '') IS NULL
        )
        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.objectKind')), '') IS NOT NULL
        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.sourceID')), '') IS NOT NULL
        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(b.Payload, '$.objectKind')), '') = NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.objectKind')), '')
        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(b.Payload, '$.sourceID')), '') = NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.sourceID')), '')
        AND (
          (
            b.EnqueueOrdinal <> 0
            AND e.EnqueueOrdinal <> 0
            AND (
              b.EnqueueOrdinal < e.EnqueueOrdinal
              OR (b.EnqueueOrdinal = e.EnqueueOrdinal AND b.Id < e.Id)
            )
          )
          OR (
            (b.EnqueueOrdinal = 0 OR e.EnqueueOrdinal = 0)
            AND (
              b.CreatedDate < e.CreatedDate
              OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence < e.DeliverySequence)
              OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence = e.DeliverySequence AND b.Id < e.Id)
            )
          )
        )
   )`

// buildClaimOrderingGuardSQL is the AND-prefixed full guard for the single-row
// claim UPDATE re-check. Returns (sql, binds).
func buildClaimOrderingGuardSQL(now interface{}) (string, []interface{}) {
	guard, binds := buildClaimOrderingGuard(now)
	return "\n   AND " + guard, binds
}

// -----------------------------------------------------------------------------
// Direction-2 (NCRM #3140): O(N) lane/delivery-head form of the claim guard for
// the HOT candidate SELECT.
//
// The two correlated NOT EXISTS branches (delivery + source-lane hot arm) are
// evaluated once per candidate row. On a burst-built deep backlog TiDB
// decorrelates them into full-table-scan hash anti-joins — O(total non-terminal
// rows) per claim, so a deep lane/delivery costs 2-3s on the hot path
// (EXPLAIN-measured). The fix computes each lane/delivery HEAD once in a grouped
// derived table and equi-joins it to e, turning the per-candidate anti-join into
// an O(N) "compute heads once, join" plan. The claimable set is byte-identical:
// by De Morgan, NOT EXISTS(cond AND (A OR B)) == NOT EXISTS(cond AND A) AND
// NOT EXISTS(cond AND B), and "no earlier non-terminal sibling by (ordinal,Id)"
// is exactly "e is the lowest-(ordinal,Id) non-terminal row in its lane", i.e.
// e IS the lane head.
//
// Only the candidate SELECT uses this form. The claim UPDATE keeps
// buildClaimOrderingGuardSQL: it is keyed on a single e.Id, so its anti-join is
// one correlated probe (O(N), not O(N^2)) and re-checks ordering atomically
// against races between the SELECT and the UPDATE.
//
// Mixed-version (EnqueueOrdinal=0) ordering and the JSON-payload fallback (rows
// missing a materialized SourceLaneKey) stay as residual NOT EXISTS predicates
// (buildClaimOrderingResidualSQL): both are empty once the backlog is upgraded,
// so they add no scan cost on the steady-state hot path.
//
// The lane-head source set carries the SAME NextAttemptAt-gated SOURCE-LANE
// blocker as branch 2: a parked-failed (future NextAttemptAt) row is not counted
// as the lane head, so its newer sibling becomes the head and is claimable —
// keeping the head-join byte-identical to the gated NOT EXISTS guard. The
// delivery-head source set stays STRICT (a parked-failed delivery head still
// blocks its tail).

// buildLaneHeadJoin joins the single claimable head of each
// (InstalledPackId, SourceLaneKey): the actionable (NextAttemptAt-gated),
// EnqueueOrdinal<>0 row with the lowest (EnqueueOrdinal, Id). ROW_NUMBER keeps
// the Id tiebreak exact when two rows share an ordinal, so only one same-source
// row is ever the head — preserving single-flight. Binds: [installedPackID, now]
// (now gates the failed-but-backing-off rows out of the head candidate set).
func buildLaneHeadJoin(installedPackID, now interface{}) (string, []interface{}) {
	sql := `
  LEFT JOIN (
    SELECT InstalledPackId, SourceLaneKey, Id AS head_id FROM (
      SELECT InstalledPackId, SourceLaneKey, Id,
             ROW_NUMBER() OVER (
               PARTITION BY InstalledPackId, SourceLaneKey
               ORDER BY EnqueueOrdinal ASC, Id ASC
             ) AS rn
        FROM _System_ConnectorEvent
       WHERE InstalledPackId = ?
         AND ` + laneHeadBlockerStatusGated + `
         AND EnqueueOrdinal <> 0
         AND NULLIF(SourceLaneKey, '') IS NOT NULL
    ) ranked
    WHERE ranked.rn = 1
  ) lh ON lh.InstalledPackId = e.InstalledPackId AND lh.SourceLaneKey = e.SourceLaneKey`
	return sql, []interface{}{installedPackID, now}
}

// buildDeliveryHeadJoin joins the lowest non-terminal DeliverySequence per
// (InstalledPackId, DeliveryId). EnqueueBatch writes a delivery atomically with a
// dense 0..N-1 DeliverySequence range, so the minimum is unique — no Id tiebreak.
// STRICT FIFO: a non-terminal (incl. parked-failed) delivery predecessor stays
// the head. Bind: [installedPackID].
func buildDeliveryHeadJoin(installedPackID interface{}) (string, []interface{}) {
	sql := `
  LEFT JOIN (
    SELECT InstalledPackId, DeliveryId, MIN(DeliverySequence) AS min_seq
      FROM _System_ConnectorEvent
     WHERE InstalledPackId = ?
       AND ` + deliveryHeadTerminalStatus + `
       AND DeliveryId IS NOT NULL
     GROUP BY InstalledPackId, DeliveryId
  ) dh ON dh.InstalledPackId = e.InstalledPackId AND dh.DeliveryId = e.DeliveryId`
	return sql, []interface{}{installedPackID}
}

// claimHeadWhereSQL is the head-equality form of the delivery branch and the
// source-lane hot arm. A candidate is delivery-claimable iff it carries no
// DeliveryId or it IS the delivery head; source-lane-claimable iff it carries no
// lane, is a mixed-version EnqueueOrdinal=0 row (handled by the residual), or it
// IS the lane head. Carries no bind.
const claimHeadWhereSQL = `
   AND (e.DeliveryId IS NULL OR e.DeliverySequence = dh.min_seq)
   AND (NULLIF(e.SourceLaneKey, '') IS NULL OR e.EnqueueOrdinal = 0 OR e.Id = lh.head_id)`

// coldResidualPredicate is the source-lane mixed-version (EnqueueOrdinal=0) arm
// of the guard, split out via De Morgan. It carries the gated SOURCE-LANE
// blocker (one `?` for `now`) so a parked-failed ordinal=0 predecessor stops
// blocking — load-bearing, because claimHeadWhereSQL lets ordinal=0 candidates
// bypass the lane-head join and rely SOLELY on this residual. It only matches
// when a legacy ordinal=0 row exists in the lane, so it is empty once the
// backlog is upgraded.
const coldResidualPredicate = `NOT EXISTS (
     SELECT 1 FROM _System_ConnectorEvent b
      WHERE b.InstalledPackId = e.InstalledPackId
        AND ` + laneBlockerStatusGated + `
        AND b.Id <> e.Id
        AND NULLIF(e.SourceLaneKey, '') IS NOT NULL
        AND b.SourceLaneKey = e.SourceLaneKey
        AND (b.EnqueueOrdinal = 0 OR e.EnqueueOrdinal = 0)
        AND (
          b.CreatedDate < e.CreatedDate
          OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence < e.DeliverySequence)
          OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence = e.DeliverySequence AND b.Id < e.Id)
        )
   )`

// buildClaimOrderingResidualSQL is the AND-tail of residual NOT EXISTS branches
// that the head joins do not cover (mixed-version ordinal=0 + JSON fallback).
// Binds in textual order: [cold now, json now].
func buildClaimOrderingResidualSQL(now interface{}) (string, []interface{}) {
	sql := "\n   AND " + coldResidualPredicate + "\n   AND " + jsonResidualPredicate
	return sql, []interface{}{now, now}
}

// buildClaimOrderingGuardExceptCurrentSQL is used by latest-by-source coalescing.
// The worker already holds currentEventID and wants to replace it with the next
// same-source row; the current row is the only earlier source-lane predecessor
// this guard may ignore. Other source-lane predecessors still block the claim.
//
// Delivery-wide predecessors remain part of the legacy guard. Only the
// versioned source-scoped handoff may relax that condition: an interleaved
// delivery A1/B1/A2 can then allow A2 to replace A1 without waiting for B1.
// Ordinary claims and every legacy latest-by-source mode remain strict.
//
// Both source-lane branches carry the gated blocker (binds: currentEventID,
// now). Binds in textual order: [b1 currentEventID, b1 now,
// b2 currentEventID, b2 now].
func buildClaimOrderingGuardExceptCurrentSQL(currentEventID, now interface{}, allowCrossSourceDelivery bool) (string, []interface{}) {
	deliveryGuard := ""
	deliveryBinds := []interface{}{}
	if !allowCrossSourceDelivery {
		deliveryGuard = `NOT EXISTS (
	     SELECT 1 FROM _System_ConnectorEvent b
	      WHERE b.InstalledPackId = e.InstalledPackId
	        AND b.Id <> ?
	        AND ` + blockerTerminalStatus + `
	        AND b.Id <> e.Id
	        AND b.DeliveryId = e.DeliveryId
	        AND b.DeliverySequence < e.DeliverySequence
	   )
	   AND `
		deliveryBinds = append(deliveryBinds, currentEventID)
	}
	sql := deliveryGuard + `NOT EXISTS (
	     SELECT 1 FROM _System_ConnectorEvent b
	      WHERE b.InstalledPackId = e.InstalledPackId
	        AND b.Id <> ?
	        AND ` + laneBlockerStatusGated + `
	        AND b.Id <> e.Id
	        AND NULLIF(e.SourceLaneKey, '') IS NOT NULL
	        AND b.SourceLaneKey = e.SourceLaneKey
	        AND (
	          (
	            b.EnqueueOrdinal <> 0
	            AND e.EnqueueOrdinal <> 0
	            AND (
	              b.EnqueueOrdinal < e.EnqueueOrdinal
	              OR (b.EnqueueOrdinal = e.EnqueueOrdinal AND b.Id < e.Id)
	            )
	          )
	          OR (
	            (b.EnqueueOrdinal = 0 OR e.EnqueueOrdinal = 0)
	            AND (
	              b.CreatedDate < e.CreatedDate
	              OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence < e.DeliverySequence)
	              OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence = e.DeliverySequence AND b.Id < e.Id)
	            )
	          )
	        )
	   )
	   AND NOT EXISTS (
	     SELECT 1 FROM _System_ConnectorEvent b
	      WHERE b.InstalledPackId = e.InstalledPackId
	        AND b.Id <> ?
	        AND ` + laneBlockerStatusGated + `
	        AND b.Id <> e.Id
	        AND (
	          NULLIF(e.SourceLaneKey, '') IS NULL
	          OR NULLIF(b.SourceLaneKey, '') IS NULL
	        )
	        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.objectKind')), '') IS NOT NULL
	        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.sourceID')), '') IS NOT NULL
	        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(b.Payload, '$.objectKind')), '') = NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.objectKind')), '')
	        AND NULLIF(JSON_UNQUOTE(JSON_EXTRACT(b.Payload, '$.sourceID')), '') = NULLIF(JSON_UNQUOTE(JSON_EXTRACT(e.Payload, '$.sourceID')), '')
	        AND (
	          (
	            b.EnqueueOrdinal <> 0
	            AND e.EnqueueOrdinal <> 0
	            AND (
	              b.EnqueueOrdinal < e.EnqueueOrdinal
	              OR (b.EnqueueOrdinal = e.EnqueueOrdinal AND b.Id < e.Id)
	            )
	          )
	          OR (
	            (b.EnqueueOrdinal = 0 OR e.EnqueueOrdinal = 0)
	            AND (
	              b.CreatedDate < e.CreatedDate
	              OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence < e.DeliverySequence)
	              OR (b.CreatedDate = e.CreatedDate AND b.DeliverySequence = e.DeliverySequence AND b.Id < e.Id)
	            )
	          )
	        )
	   )`
	return sql, append(deliveryBinds, currentEventID, now, currentEventID, now)
}