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"
)

// WorkItemCoalescedSiblingClaim pairs immutable Event input with the WorkItem
// lease that owns its execution.
type WorkItemCoalescedSiblingClaim struct {
	Event *ConnectorEvent
	Work  *connector_workqueue.WorkItem
}

// ClaimNextCoalescedExecutorWorkItemSibling atomically completes the current
// ordinary WorkItem and claims its latest_by_source successor. The successor
// may be a fresh Event or an already-frozen retry; any one-time Event ownership
// transfer is part of the same WorkItem epoch CAS.
func (s *Service) ClaimNextCoalescedExecutorWorkItemSibling(
	ctx context.Context,
	currentEventID, currentWorkID, currentLeaseToken, workerID,
	ownerID, fencingToken, nextLeaseToken string,
	generation int64,
	leaseUntil, now time.Time,
) (*WorkItemCoalescedSiblingClaim, bool, error) {
	return s.claimNextCoalescedExecutorWorkItemSibling(
		ctx, currentEventID, currentWorkID, currentLeaseToken, workerID,
		ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now, false,
	)
}

// ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped applies the same
// WorkItem-owned handoff while allowing the route's source-scoped delivery
// semantics.
func (s *Service) ClaimNextCoalescedExecutorWorkItemSiblingSourceScoped(
	ctx context.Context,
	currentEventID, currentWorkID, currentLeaseToken, workerID,
	ownerID, fencingToken, nextLeaseToken string,
	generation int64,
	leaseUntil, now time.Time,
) (*WorkItemCoalescedSiblingClaim, bool, error) {
	return s.claimNextCoalescedExecutorWorkItemSibling(
		ctx, currentEventID, currentWorkID, currentLeaseToken, workerID,
		ownerID, fencingToken, nextLeaseToken, generation, leaseUntil, now, true,
	)
}

func (s *Service) claimNextCoalescedExecutorWorkItemSibling(
	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 == "" ||
		generation < 0 || now.IsZero() || !leaseUntil.After(now) {
		return nil, false, fmt.Errorf("connector events: invalid WorkItem coalesced claim")
	}
	if s.workqueue == nil {
		return nil, false, fmt.Errorf("connector events: WorkItem repository is unavailable")
	}

	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
	err = tx.QueryRowContext(ctx, `
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 ExecutionOwner = ?
   AND Status = ?
 FOR UPDATE`,
		currentEventID, connector_workqueue.EventExecutionOwnerWorkItem,
		connector_workqueue.EventWorkItemCompatibilityFenceStatus,
	).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.nextCoalescibleWorkItemSibling(
		ctx, tx, currentEventID, current, allowCrossSourceDelivery, generation,
	)
	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.CompleteExecutorWorkItemForCoalescing(
		ctx, currentWorkID, currentEventID, currentLeaseToken, generation, now,
	)
	if err != nil {
		return nil, false, err
	}
	if !completedCurrent {
		return nil, false, ErrLeaseLost
	}

	claimCtx := connector_workqueue.WithClaimWorkTypes(ctx, []string{
		connector_workqueue.WorkTypeConnectorEvent,
		connector_workqueue.WorkTypeWorkflowActionApply,
	})
	nextWork, claimedWork, err := workRepo.ClaimWorkItemByEventForCoalescing(
		claimCtx,
		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,
		},
		allowCrossSourceDelivery,
	)
	if err != nil {
		return nil, false, err
	}
	if !claimedWork {
		// Completion and successor claim are one handoff. Rollback restores the
		// current lease so the caller can still execute the current snapshot.
		return nil, false, nil
	}

	event, 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
	event.ClaimToken = nextLeaseToken
	return &WorkItemCoalescedSiblingClaim{Event: event, Work: nextWork}, true, nil
}