package connector_events

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

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

const (
	workItemFailureClassInboundExecution = connector_workqueue.FailureClassInboundExecution
	workItemFailureSummaryInbound        = connector_workqueue.FailureSummaryInbound
)

func workItemFailureRetryDelays(threshold int) []time.Duration {
	if threshold <= 0 {
		return nil
	}
	delays := make([]time.Duration, threshold)
	for i := range delays {
		delays[i] = claimpolicy.FailureRetryDelay(i + 1)
	}
	return delays
}

type workItemEventQueryExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

// MarkEventProcessingForWorkItem mirrors a WorkItem lease onto the legacy event
// row so explicit work_item mode is safe to roll back to legacy_event mode.
func (s *Service) MarkEventProcessingForWorkItem(ctx context.Context, eventID, workerID, leaseToken string, claimedAt, now time.Time) error {
	eventID = strings.TrimSpace(eventID)
	workerID = strings.TrimSpace(workerID)
	leaseToken = strings.TrimSpace(leaseToken)
	if eventID == "" || workerID == "" || leaseToken == "" || claimedAt.IsZero() || now.IsZero() {
		return fmt.Errorf("connector events: invalid WorkItem event processing bridge")
	}
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = ?,
       ClaimedBy = ?,
       ClaimedAt = ?,
       ClaimToken = ?,
       NextAttemptAt = NULL,
       LastModifiedDate = ?
 WHERE Id = ?
   AND (
       (Status = ? AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?))
       OR (Status = ? AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?))
       OR (Status = ? AND LastModifiedDate < ?)
   )`,
		statusProcessing, workerID, claimedAt, leaseToken, now, eventID,
		statusPending, now, statusFailed, now, statusProcessing, now.Add(-StaleProcessingSeconds*time.Second))
	if err != nil {
		return err
	}
	ok, err := workItemBridgeRowsAffected(res)
	if err != nil {
		return err
	}
	if ok {
		return nil
	}
	_, status, loadErr := s.loadNamespaceAndStatus(ctx, eventID)
	if loadErr != nil {
		return loadErr
	}
	if status == statusProcessed || status == statusSkipped || status == statusDead {
		return ErrEventAlreadyTerminal
	}
	return ErrLeaseLost
}

func (s *Service) ClaimNextCoalescedWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*WorkItemCoalescedSiblingClaim, bool, error) {
	return s.claimNextCoalescedWorkItemSibling(ctx, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now, false)
}

// ClaimNextCoalescedWorkItemSiblingSourceScoped is the WorkItem equivalent of
// the versioned source-scoped event handoff.
func (s *Service) ClaimNextCoalescedWorkItemSiblingSourceScoped(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time) (*WorkItemCoalescedSiblingClaim, bool, error) {
	return s.claimNextCoalescedWorkItemSibling(ctx, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now, true)
}

func (s *Service) claimNextCoalescedWorkItemSibling(ctx context.Context, currentEventID, currentWorkID, currentLeaseToken, workerID, ownerID, fencingToken, nextLeaseToken string, generation int64, leaseUntil, now time.Time, allowCrossSourceDelivery bool) (*WorkItemCoalescedSiblingClaim, bool, error) {
	currentEventID = strings.TrimSpace(currentEventID)
	currentWorkID = strings.TrimSpace(currentWorkID)
	currentLeaseToken = strings.TrimSpace(currentLeaseToken)
	workerID = strings.TrimSpace(workerID)
	ownerID = strings.TrimSpace(ownerID)
	fencingToken = strings.TrimSpace(fencingToken)
	nextLeaseToken = strings.TrimSpace(nextLeaseToken)
	if currentEventID == "" || currentWorkID == "" || currentLeaseToken == "" || workerID == "" || ownerID == "" || fencingToken == "" || nextLeaseToken == "" || !leaseUntil.After(now) || now.IsZero() {
		return nil, false, fmt.Errorf("connector events: invalid WorkItem coalesced claim bridge")
	}

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

	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
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`
	currentArgs := []any{currentEventID, statusProcessing, currentLeaseToken}
	err = tx.QueryRowContext(ctx, currentQuery+`
 FOR UPDATE`, currentArgs...).Scan(
		&current.installedPackID,
		&current.eventKind,
		&current.routeVersion,
		&current.companionPackInstallID,
		&current.companionFlowVersionID,
		&current.sourceLaneKey,
		&current.objectKind,
		&current.sourceID,
		&current.deliverySequence,
		&current.enqueueOrdinal,
		&current.createdDate,
	)
	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
	}

	workRepo := s.workqueue.WithTx(tx)
	completedCurrent, err := workRepo.CompleteWorkItem(ctx, currentWorkID, currentLeaseToken, connector_workqueue.WorkStateProcessed, now)
	if err != nil {
		return nil, false, err
	}
	if !completedCurrent {
		return nil, false, ErrLeaseLost
	}
	nextWork, claimedWork, err := workRepo.ClaimWorkItemByEvent(ctx, connector_workqueue.ClaimWorkItemByEventInput{
		EventID:         next.id,
		WorkType:        workTypeForEventKind(next.eventKind),
		InstalledPackID: current.installedPackID,
		Generation:      generation,
		OwnerID:         ownerID,
		FencingToken:    fencingToken,
		LeaseToken:      nextLeaseToken,
		LeaseUntil:      leaseUntil,
		Now:             now,
	})
	if err != nil {
		return nil, false, err
	}
	if !claimedWork {
		// Completion and successor claim are one handoff. A lost successor CAS
		// leaves the current lease authoritative so the caller can execute it;
		// the deferred rollback restores the completion above.
		return nil, false, nil
	}

	res, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = ?,
       ClaimedBy = ?,
       ClaimedAt = ?,
       ClaimToken = ?,
       LastModifiedDate = ?
 WHERE Id = ?
   AND (
       (Status = ? AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?))
       OR (Status = ? AND (NextAttemptAt IS NULL OR NextAttemptAt <= ?))
   )`,
		statusProcessing, workerID, now, nextLeaseToken, now,
		next.id, statusPending, now, statusFailed, now)
	if err != nil {
		return nil, false, err
	}
	updatedNext, err := workItemBridgeRowsAffected(res)
	if err != nil {
		return nil, false, err
	}
	if !updatedNext {
		return nil, false, nil
	}

	updatedCurrent, err := s.markCurrentEventProcessedForWorkItemCoalesce(ctx, tx, currentEventID, currentLeaseToken, now)
	if err != nil {
		return nil, false, err
	}
	if !updatedCurrent {
		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 = nextLeaseToken
	return &WorkItemCoalescedSiblingClaim{Event: ev, Work: nextWork}, true, nil
}

func (s *Service) markCurrentEventProcessedForWorkItemCoalesce(ctx context.Context, tx *sql.Tx, eventID, leaseToken string, now time.Time) (bool, error) {
	res, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = ?,
       ProcessedAt = ?,
       LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		statusProcessed, now, now, eventID, statusProcessing, leaseToken)
	if err != nil {
		return false, err
	}
	return workItemBridgeRowsAffected(res)
}

// MarkEventTerminalForWorkItem mirrors WorkItem terminal state onto the legacy
// event row before the WorkItem is completed, preventing rollback re-execution.
func (s *Service) MarkEventTerminalForWorkItem(ctx context.Context, eventID, leaseToken, terminalStatus, errorMsg string, now time.Time) error {
	eventID = strings.TrimSpace(eventID)
	leaseToken = strings.TrimSpace(leaseToken)
	terminalStatus = strings.TrimSpace(terminalStatus)
	if err := validateWorkItemTerminalInput(eventID, leaseToken, terminalStatus, now); err != nil {
		return err
	}
	return markEventTerminalForWorkItem(ctx, s.db, eventID, leaseToken, terminalStatus, errorMsg, now)
}

// CompleteEventForWorkItem completes the WorkItem generation under its lease
// token. Event remains immutable execution input for WorkItem-owned rows.
func (s *Service) CompleteEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, terminalStatus, errorMsg string, now time.Time) (bool, error) {
	eventID = strings.TrimSpace(eventID)
	workID = strings.TrimSpace(workID)
	leaseToken = strings.TrimSpace(leaseToken)
	terminalStatus = strings.TrimSpace(terminalStatus)
	if err := validateWorkItemTerminalInput(eventID, leaseToken, terminalStatus, now); err != nil {
		return false, err
	}
	if s.workqueue == nil {
		return false, fmt.Errorf("connector events: WorkItem terminal bridge is unavailable")
	}
	return s.workqueue.CompleteWorkItem(ctx, workID, leaseToken, terminalStatus, now)
}

// CompleteSourceProjectionEventForWorkItem is retained for callers that
// explicitly need the durable source-projection handoff contract.
func (s *Service) CompleteSourceProjectionEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, terminalStatus, errorMsg string, now time.Time) (bool, error) {
	if !connector_workqueue.IsSourceProjectionWorkID(strings.TrimSpace(workID)) {
		return false, fmt.Errorf("connector events: invalid source-projection WorkItem terminal bridge")
	}
	return s.CompleteEventForWorkItem(ctx, eventID, workID, leaseToken, terminalStatus, errorMsg, now)
}

// FailEventForWorkItem consumes the WorkItem-owned failure budget. The Event
// row is immutable on this repository-backed path.
func (s *Service) FailEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, errorMsg string, now time.Time) (*connector_workqueue.FailWorkItemResult, bool, error) {
	eventID = strings.TrimSpace(eventID)
	workID = strings.TrimSpace(workID)
	leaseToken = strings.TrimSpace(leaseToken)
	if eventID == "" || workID == "" || leaseToken == "" || now.IsZero() {
		return nil, false, fmt.Errorf("connector events: invalid WorkItem failure bridge")
	}
	if s.workqueue == nil {
		return nil, false, fmt.Errorf("connector events: WorkItem failure bridge is unavailable")
	}
	return s.workqueue.FailWorkItem(ctx, connector_workqueue.FailWorkItemInput{
		WorkID:         workID,
		EventID:        eventID,
		LeaseToken:     leaseToken,
		FailureClass:   workItemFailureClassInboundExecution,
		FailureSummary: workItemFailureSummaryInbound,
		RetryThreshold: s.retryThreshold,
		RetryDelays:    workItemFailureRetryDelays(s.retryThreshold),
		Now:            now,
	})
}

func validateWorkItemTerminalInput(eventID, leaseToken, terminalStatus string, now time.Time) error {
	if eventID == "" || leaseToken == "" || now.IsZero() {
		return fmt.Errorf("connector events: invalid WorkItem event terminal bridge")
	}
	if terminalStatus != statusProcessed && terminalStatus != statusSkipped && terminalStatus != statusDead {
		return fmt.Errorf("connector events: invalid WorkItem terminal status %q", terminalStatus)
	}
	return nil
}

func markEventTerminalForWorkItem(ctx context.Context, q workItemEventQueryExecer, eventID, leaseToken, terminalStatus, errorMsg string, now time.Time) error {
	processedAt := any(nil)
	if terminalStatus == statusProcessed || terminalStatus == statusSkipped {
		processedAt = now
	}
	lastError := any(nil)
	lastErrorAt := any(nil)
	if terminalStatus == statusDead && strings.TrimSpace(errorMsg) != "" {
		lastError = errorMsg
		lastErrorAt = now
	}
	res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = ?,
       ProcessedAt = ?,
       LastError = ?,
       LastErrorAt = ?,
       NextAttemptAt = NULL,
       LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		terminalStatus, processedAt, lastError, lastErrorAt, now,
		eventID, statusProcessing, leaseToken)
	if err != nil {
		return err
	}
	ok, err := workItemBridgeRowsAffected(res)
	if err != nil {
		return err
	}
	if ok {
		return nil
	}
	var status string
	if err := q.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&status); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrEventNotFound
		}
		return err
	}
	if status == statusProcessed || status == statusSkipped || status == statusDead {
		return ErrEventAlreadyTerminal
	}
	return ErrLeaseLost
}

func (s *Service) ReleaseEventForWorkItemRetry(ctx context.Context, eventID, leaseToken, errorMsg string, runAfter, now time.Time, retryCount int) error {
	eventID = strings.TrimSpace(eventID)
	leaseToken = strings.TrimSpace(leaseToken)
	if retryCount < 0 {
		retryCount = 0
	}
	if err := validateWorkItemRetryInput(eventID, leaseToken, runAfter, now); err != nil {
		return err
	}
	return releaseEventForWorkItemRetry(ctx, s.db, eventID, leaseToken, errorMsg, runAfter, now, retryCount)
}

// RetrySourceProjectionEventForWorkItem retains the compatibility method shape
// while using the WorkItem lease as the sole mutable retry authority. Event is
// immutable input after ownership transfer.
func (s *Service) RetrySourceProjectionEventForWorkItem(ctx context.Context, eventID, workID, leaseToken, errorMsg string, runAfter, now time.Time, retryCount int) (bool, error) {
	eventID = strings.TrimSpace(eventID)
	workID = strings.TrimSpace(workID)
	leaseToken = strings.TrimSpace(leaseToken)
	if retryCount < 0 {
		retryCount = 0
	}
	if err := validateWorkItemRetryInput(eventID, leaseToken, runAfter, now); err != nil {
		return false, err
	}
	if !connector_workqueue.IsSourceProjectionWorkID(workID) || s.workqueue == nil {
		return false, fmt.Errorf("connector events: invalid source-projection WorkItem retry bridge")
	}
	return s.workqueue.RetryWorkItem(ctx, workID, leaseToken, runAfter, now)
}

func validateWorkItemRetryInput(eventID, leaseToken string, runAfter, now time.Time) error {
	if eventID == "" || leaseToken == "" || runAfter.IsZero() || now.IsZero() {
		return fmt.Errorf("connector events: invalid WorkItem event retry bridge")
	}
	return nil
}

func releaseEventForWorkItemRetry(ctx context.Context, q workItemEventQueryExecer, eventID, leaseToken, errorMsg string, runAfter, now time.Time, retryCount int) error {
	res, err := q.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET Status = ?,
       RetryCount = ?,
       LastError = ?,
       LastErrorAt = ?,
       NextAttemptAt = ?,
       ClaimedBy = NULL,
       ClaimedAt = NULL,
       ClaimToken = NULL,
       LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		statusFailed, retryCount, errorMsg, now, runAfter, now,
		eventID, statusProcessing, leaseToken)
	if err != nil {
		return err
	}
	ok, err := workItemBridgeRowsAffected(res)
	if err != nil {
		return err
	}
	if ok {
		return nil
	}
	var status string
	if err := q.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, eventID).Scan(&status); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return ErrEventNotFound
		}
		return err
	}
	if status == statusProcessed || status == statusSkipped || status == statusDead {
		return ErrEventAlreadyTerminal
	}
	return ErrLeaseLost
}

func (s *Service) RenewEventLeaseForWorkItem(ctx context.Context, eventID, leaseToken string, now time.Time) error {
	eventID = strings.TrimSpace(eventID)
	leaseToken = strings.TrimSpace(leaseToken)
	if eventID == "" || leaseToken == "" || now.IsZero() {
		return fmt.Errorf("connector events: invalid WorkItem event renew bridge")
	}
	res, err := s.db.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ClaimToken = ?`,
		now, eventID, statusProcessing, leaseToken)
	if err != nil {
		return err
	}
	ok, err := workItemBridgeRowsAffected(res)
	if err != nil {
		return err
	}
	if !ok {
		return ErrLeaseLost
	}
	return nil
}

func workItemBridgeRowsAffected(res sql.Result) (bool, error) {
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}