package connector_events

import (
	"context"
	"database/sql"
	"strings"
	"time"
)

// BatchClaimRequest describes one event-to-worker lease assignment for
// ClaimBatchByID.
type BatchClaimRequest struct {
	EventID  string
	WorkerID string
}

// BatchClaimResult reports a row that won the batch CAS.
type BatchClaimResult struct {
	EventID    string
	WorkerID   string
	ClaimToken string
	ClaimedAt  time.Time
}

// PeekBatch returns up to n pending/failed/stale rows for installedPackID in
// (CreatedDate ASC, DeliverySequence ASC, Id ASC) order WITHOUT claiming them.
// It intentionally does not apply the global ordering guard: batch_collection
// workers need to see later same-delivery siblings after claiming the anchor.
// It does exclude source-projection lanes owned by a durable WorkItem, because
// no legacy Event claimant may bypass that Event-to-WorkItem bridge.
// ClaimByID remains the CAS arbiter and still enforces ordering.
//
// n <= 0 → default 1.
//
// Does not gate on pack activation state — same guarantee as ClaimNext.
func (s *Service) PeekBatch(ctx context.Context, installedPackID string, n int) ([]*ConnectorEvent, error) {
	return s.peekBatchImpl(ctx, installedPackID, nil, n, false)
}

func (s *Service) PeekBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*ConnectorEvent, error) {
	return s.peekBatchImpl(ctx, installedPackID, queueClasses, n, false)
}

// PeekClaimableBatch returns up to n rows that are actually claimable under the
// same delivery/source-lane head rules used by ClaimNext. Unlike PeekBatch, this
// is not an observational batch-collection helper: callers use it immediately
// before ClaimBatchByID, so returning non-head rows only creates CAS misses.
func (s *Service) PeekClaimableBatch(ctx context.Context, installedPackID string, n int) ([]*ConnectorEvent, error) {
	return s.peekClaimableBatchImpl(ctx, installedPackID, nil, n, false)
}

func (s *Service) PeekClaimableBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*ConnectorEvent, error) {
	return s.peekClaimableBatchImpl(ctx, installedPackID, queueClasses, n, false)
}

// PeekStaleBatch / PeekStaleBatchForClasses are the stale-reclaim-only peek used
// by the single-worker drain path under a limp permit: they surface ONLY
// stale-processing rows, so the worker's subsequent ClaimByID re-leases a
// stranded row instead of admitting a fresh pending/failed row past the cap.
func (s *Service) PeekStaleBatch(ctx context.Context, installedPackID string, n int) ([]*ConnectorEvent, error) {
	return s.peekBatchImpl(ctx, installedPackID, nil, n, true)
}

func (s *Service) PeekStaleBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*ConnectorEvent, error) {
	return s.peekBatchImpl(ctx, installedPackID, queueClasses, n, true)
}

func (s *Service) PeekClaimableStaleBatch(ctx context.Context, installedPackID string, n int) ([]*ConnectorEvent, error) {
	return s.peekClaimableBatchImpl(ctx, installedPackID, nil, n, true)
}

func (s *Service) PeekClaimableStaleBatchForClasses(ctx context.Context, installedPackID string, queueClasses []string, n int) ([]*ConnectorEvent, error) {
	return s.peekClaimableBatchImpl(ctx, installedPackID, queueClasses, n, true)
}

func (s *Service) peekBatchImpl(ctx context.Context, installedPackID string, queueClasses []string, n int, staleOnly bool) ([]*ConnectorEvent, error) {
	if n <= 0 {
		n = 1
	}
	queueClassSQL, queueClassArgs, err := queueClassFilterSQL("e", queueClasses)
	if err != nil {
		return nil, err
	}
	now := s.now()
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)
	// staleOnly narrows the peek to the stale-processing arm (the limp path), so a
	// degraded single-worker tick surfaces only stranded rows to reclaim.
	peekStatusSQL := claimCandidateStatusSQL
	peekStatusArgs := []interface{}{now, now, staleCutoff}
	if staleOnly {
		peekStatusSQL = claimStaleOnlyStatusSQL
		peekStatusArgs = []interface{}{staleCutoff}
	}
	baseQuery := `
SELECT e.Id, e.Namespace, e.InstalledPackId, e.EventKind, e.IdempotencyKey, e.Status,
       e.QueueClass, e.Payload, e.RetryCount, e.LastError, e.LastErrorAt,
       e.ClaimedBy, e.ClaimedAt, e.ClaimToken, e.ProcessedAt, e.ReplayedAt, e.ReplayOfId,
       e.RouteVersion, e.CompanionPackInstallId, e.CompanionFlowVersionId,
       e.SourceLaneKey, e.DeliveryId, e.DeliverySequence, e.CreatedDate, e.LastModifiedDate
 FROM _System_ConnectorEvent e
 WHERE e.InstalledPackId = ?
   AND ` + peekStatusSQL + `
` + queueClassSQL
	orderSQL := `
		 ORDER BY e.CreatedDate ASC, e.DeliverySequence ASC, e.Id ASC
		 LIMIT ?`
	var rows *sql.Rows
	mode := legacyEventWorkItemGuardFull
	for {
		workItemGuardSQL, workItemGuardBinds := s.legacyEventWorkItemGuardSQL("e", mode)
		peekArgs := append([]interface{}{installedPackID}, peekStatusArgs...)
		peekArgs = append(peekArgs, queueClassArgs...)
		peekArgs = append(peekArgs, workItemGuardBinds...)
		peekArgs = append(peekArgs, n)
		rows, err = s.db.QueryContext(ctx, baseQuery+workItemGuardSQL+orderSQL, peekArgs...)
		nextMode, retry := nextLegacyEventWorkItemGuardMode(mode, err)
		if retry {
			mode = nextMode
			continue
		}
		break
	}
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	evs, err := scanConnectorEvents(rows)
	if err != nil {
		return nil, err
	}
	result := make([]*ConnectorEvent, len(evs))
	for i := range evs {
		ev := evs[i]
		result[i] = &ev
	}
	return result, nil
}

func (s *Service) peekClaimableBatchImpl(ctx context.Context, installedPackID string, queueClasses []string, n int, staleOnly bool) ([]*ConnectorEvent, error) {
	if n <= 0 {
		n = 1
	}
	queueClassSQL, queueClassArgs, err := queueClassFilterSQL("e", queueClasses)
	if err != nil {
		return nil, err
	}
	now := s.now()
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)
	candidateStatusSQL := claimCandidateStatusSQL
	candidateStatusArgs := []interface{}{now, now, staleCutoff}
	if staleOnly {
		candidateStatusSQL = claimStaleOnlyStatusSQL
		candidateStatusArgs = []interface{}{staleCutoff}
	}
	laneHeadJoinSQL, laneHeadBinds := buildLaneHeadJoin(installedPackID, now)
	deliveryHeadJoinSQL, deliveryHeadBinds := buildDeliveryHeadJoin(installedPackID)
	residualSQL, residualBinds := buildClaimOrderingResidualSQL(now)

	baseQuery := `
SELECT e.Id, e.Namespace, e.InstalledPackId, e.EventKind, e.IdempotencyKey, e.Status,
       e.QueueClass, e.Payload, e.RetryCount, e.LastError, e.LastErrorAt,
       e.ClaimedBy, e.ClaimedAt, e.ClaimToken, e.ProcessedAt, e.ReplayedAt, e.ReplayOfId,
       e.RouteVersion, e.CompanionPackInstallId, e.CompanionFlowVersionId,
       e.SourceLaneKey, e.DeliveryId, e.DeliverySequence, e.CreatedDate, e.LastModifiedDate
 FROM _System_ConnectorEvent e` + laneHeadJoinSQL + deliveryHeadJoinSQL + `
 WHERE e.InstalledPackId = ?
   AND ` + candidateStatusSQL + `
` + queueClassSQL + claimHeadWhereSQL + residualSQL
	orderSQL := `
 ORDER BY e.CreatedDate ASC, e.DeliverySequence ASC, e.Id ASC
 LIMIT ?`
	var rows *sql.Rows
	mode := legacyEventWorkItemGuardFull
	for {
		workItemGuardSQL, workItemGuardBinds := s.legacyEventWorkItemGuardSQL("e", mode)
		args := append([]interface{}{}, laneHeadBinds...)
		args = append(args, deliveryHeadBinds...)
		args = append(args, installedPackID)
		args = append(args, candidateStatusArgs...)
		args = append(args, queueClassArgs...)
		args = append(args, residualBinds...)
		args = append(args, workItemGuardBinds...)
		args = append(args, n)
		rows, err = s.db.QueryContext(ctx, baseQuery+workItemGuardSQL+orderSQL, args...)
		nextMode, retry := nextLegacyEventWorkItemGuardMode(mode, err)
		if retry {
			mode = nextMode
			continue
		}
		break
	}
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	evs, err := scanConnectorEvents(rows)
	if err != nil {
		return nil, err
	}
	result := make([]*ConnectorEvent, len(evs))
	for i := range evs {
		ev := evs[i]
		result[i] = &ev
	}
	return result, nil
}

// ClaimByID atomically claims one specific row identified by eventID.
// It runs a CAS UPDATE keyed on a known Id and returns the fresh token plus the
// timestamp written to ClaimedAt:
//
//	UPDATE ... SET Status='processing', ClaimedBy=?, ClaimedAt=?, ClaimToken=<fresh>,
//	               LastModifiedDate=?
//	         WHERE Id=?
//	           AND (Status='pending'
//	                OR (Status='failed' AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?))
//	                OR (Status='processing' AND LastModifiedDate < ?))
//	           AND NOT EXISTS (earlier sibling in same DeliveryId OR same source lane NOT terminal)
//
// The NOT EXISTS guard mirrors ClaimNext's UPDATE-side guard: a claim is
// rejected if a non-terminal earlier sibling exists either in the same
// DeliveryId or in the same queue-envelope lane (objectKind + sourceID),
// closing the TOCTOU window between PeekBatch and ClaimByID.
//
// The deliveryID / deliverySequence arguments remain on the method signature for
// interface compatibility with existing drain callers and tests. The SQL guard is
// now fully correlated against the target row `e`, so those parameters are no
// longer the sole ordering inputs.
//
// RowsAffected==1 → (freshToken, claimedAt, true, nil). The caller holds the
// full row from PeekBatch; it stamps the returned token and claim timestamp
// onto a local copy. No loadByID is issued — eliminating the "CAS won but load
// errored → already-claimed row stranded" class of failure.
// RowsAffected==0 → ("", time.Time{}, false, nil): race lost or ordering guard fired; caller skips silently.
// CAS exec error → ("", time.Time{}, false, err): claim did not cleanly succeed; caller skips.
//
// Mirrors ClaimNext's CAS semantics exactly, just keyed on eventID.
func (s *Service) ClaimByID(ctx context.Context, eventID, workerID, _ string, _ sql.NullString, _ int) (string, time.Time, bool, error) {
	now := s.now()
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)
	freshToken := s.idgen()

	guardSQL, guardBinds := buildClaimOrderingGuardSQL(now)
	updateArgs := append([]interface{}{statusProcessing, workerID, now, freshToken, now, eventID, now, now, staleCutoff}, guardBinds...)
	query := `
UPDATE _System_ConnectorEvent e
   SET Status = ?,
       ClaimedBy = ?,
       ClaimedAt = ?,
       ClaimToken = ?,
       LastModifiedDate = ?
 WHERE e.Id = ?
   AND (
     (e.Status = 'pending' AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
     OR (e.Status = 'failed' AND (e.NextAttemptAt IS NULL OR e.NextAttemptAt <= ?))
     OR (e.Status = 'processing' AND e.LastModifiedDate < ?)
   )
		` + guardSQL
	res, err := s.execLegacyEventMutation(ctx, s.db, query, updateArgs...)
	if err != nil {
		recordConnectorEventClaim("by_id", "error")
		return "", time.Time{}, false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		recordConnectorEventClaim("by_id", "error")
		return "", time.Time{}, false, err
	}
	if n == 0 {
		// Lost race — another worker claimed it, it moved to a terminal state,
		// or the NOT EXISTS ordering guard fired (earlier sibling not yet terminal).
		recordConnectorEventClaim("by_id", "miss")
		return "", time.Time{}, false, nil
	}
	// Won the CAS. Return the fresh token so the caller can stamp it onto the
	// peeked row. No loadByID — the worker already holds the full row from
	// PeekBatch (payload/RouteVersion/Companion are immutable post-enqueue).
	recordConnectorEventClaim("by_id", "claimed")
	return freshToken, now, true, nil
}

// ClaimBatchByID atomically claims multiple known rows with the same per-row
// CAS and ordering guard as ClaimByID. It is used by the central inbound
// scheduler after a batch peek has selected candidate rows for ready workers.
//
// Each request gets its own WorkerID and fresh ClaimToken. Rows that fail the
// status predicate or ordering guard are simply absent from the returned map.
// The method does not gate on pack activation state, matching ClaimByID and
// ClaimNext: once an event exists, drain/finalize must continue through pack
// lifecycle transitions.
func (s *Service) ClaimBatchByID(ctx context.Context, requests []BatchClaimRequest) (map[string]BatchClaimResult, error) {
	prepared := make([]batchClaimPrepared, 0, len(requests))
	seen := make(map[string]struct{}, len(requests))
	for _, req := range requests {
		eventID := strings.TrimSpace(req.EventID)
		workerID := strings.TrimSpace(req.WorkerID)
		if eventID == "" || workerID == "" {
			continue
		}
		if _, ok := seen[eventID]; ok {
			continue
		}
		seen[eventID] = struct{}{}
		prepared = append(prepared, batchClaimPrepared{
			eventID:  eventID,
			workerID: workerID,
			token:    s.idgen(),
		})
	}
	results := make(map[string]BatchClaimResult, len(prepared))
	if len(prepared) == 0 {
		return results, nil
	}

	now := s.now()
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)
	guardSQL, guardBinds := buildClaimOrderingGuardSQL(now)

	workerCaseSQL, workerCaseArgs := batchClaimCaseSQL(prepared, func(row batchClaimPrepared) string {
		return row.workerID
	}, "ClaimedBy")
	tokenCaseSQL, tokenCaseArgs := batchClaimCaseSQL(prepared, func(row batchClaimPrepared) string {
		return row.token
	}, "ClaimToken")
	eventIDs := make([]interface{}, 0, len(prepared))
	for _, row := range prepared {
		eventIDs = append(eventIDs, row.eventID)
	}

	updateArgs := make([]interface{}, 0, 3+len(workerCaseArgs)+len(tokenCaseArgs)+len(eventIDs)+len(guardBinds))
	updateArgs = append(updateArgs, statusProcessing)
	updateArgs = append(updateArgs, workerCaseArgs...)
	updateArgs = append(updateArgs, now)
	updateArgs = append(updateArgs, tokenCaseArgs...)
	updateArgs = append(updateArgs, now)
	updateArgs = append(updateArgs, eventIDs...)
	updateArgs = append(updateArgs, now, now, staleCutoff)
	updateArgs = append(updateArgs, guardBinds...)
	query := `
UPDATE _System_ConnectorEvent e
   SET Status = ?,
       ClaimedBy = ` + workerCaseSQL + `,
       ClaimedAt = ?,
       ClaimToken = ` + tokenCaseSQL + `,
       LastModifiedDate = ?
 WHERE e.Id IN (` + sqlPlaceholders(len(prepared)) + `)
	   AND ` + claimCandidateStatusSQL + guardSQL
	res, err := s.execLegacyEventMutation(ctx, s.db, query, updateArgs...)
	if err != nil {
		recordConnectorEventClaim("batch_by_id", "error")
		return nil, err
	}
	affected, err := res.RowsAffected()
	if err != nil {
		recordConnectorEventClaim("batch_by_id", "error")
		return nil, err
	}
	if affected <= 0 {
		for range prepared {
			recordConnectorEventClaim("batch_by_id", "miss")
		}
		return results, nil
	}
	if int(affected) == len(prepared) {
		for _, row := range prepared {
			results[row.eventID] = BatchClaimResult{
				EventID:    row.eventID,
				WorkerID:   row.workerID,
				ClaimToken: row.token,
				ClaimedAt:  now,
			}
			recordConnectorEventClaim("batch_by_id", "claimed")
		}
		return results, nil
	}

	partial, err := s.loadBatchClaimWinners(ctx, prepared)
	if err != nil {
		recordConnectorEventClaim("batch_by_id", "error")
		return nil, err
	}
	for eventID, result := range partial {
		results[eventID] = result
		recordConnectorEventClaim("batch_by_id", "claimed")
	}
	for i := 0; i < len(prepared)-len(partial); i++ {
		recordConnectorEventClaim("batch_by_id", "miss")
	}
	return results, nil
}

type batchClaimPrepared struct {
	eventID  string
	workerID string
	token    string
}

func batchClaimCaseSQL(rows []batchClaimPrepared, value func(batchClaimPrepared) string, elseColumn string) (string, []interface{}) {
	var builder strings.Builder
	builder.WriteString("CASE e.Id")
	args := make([]interface{}, 0, len(rows)*2)
	for _, row := range rows {
		builder.WriteString(" WHEN ? THEN ?")
		args = append(args, row.eventID, value(row))
	}
	builder.WriteString(" ELSE ")
	builder.WriteString(elseColumn)
	builder.WriteString(" END")
	return builder.String(), args
}

func (s *Service) loadBatchClaimWinners(ctx context.Context, prepared []batchClaimPrepared) (map[string]BatchClaimResult, error) {
	results := make(map[string]BatchClaimResult, len(prepared))
	if len(prepared) == 0 {
		return results, nil
	}
	eventIDs := make([]interface{}, 0, len(prepared))
	tokens := make([]interface{}, 0, len(prepared))
	byID := make(map[string]batchClaimPrepared, len(prepared))
	for _, row := range prepared {
		eventIDs = append(eventIDs, row.eventID)
		tokens = append(tokens, row.token)
		byID[row.eventID] = row
	}
	args := append(eventIDs, tokens...)
	rows, err := s.db.QueryContext(ctx, `
SELECT Id, ClaimedBy, ClaimToken, ClaimedAt
  FROM _System_ConnectorEvent
 WHERE Id IN (`+sqlPlaceholders(len(prepared))+`)
   AND ClaimToken IN (`+sqlPlaceholders(len(prepared))+`)`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var eventID, workerID, token string
		var claimedAt time.Time
		if err := rows.Scan(&eventID, &workerID, &token, &claimedAt); err != nil {
			return nil, err
		}
		expected, ok := byID[eventID]
		if !ok || expected.workerID != workerID || expected.token != token {
			continue
		}
		results[eventID] = BatchClaimResult{
			EventID:    eventID,
			WorkerID:   workerID,
			ClaimToken: token,
			ClaimedAt:  claimedAt,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return results, nil
}