package connector_events

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

	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connector_workqueue "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
)

type DeadWorkItem = connector_workqueue.DeadWorkItem

// GetDeadWorkItem loads the effective executor state for one Event. A WorkItem
// generation wins after it has acquired a WorkItem lease at least once; a
// zero-attempt compatibility mirror remains on the legacy Event read path.
func (s *Service) GetDeadWorkItem(ctx context.Context, perms PermissionChecker, eventID string) (*DeadWorkItem, error) {
	if !perms.Has(string(constants.PermReplayConnectorEvents)) {
		return nil, ErrPermissionDenied
	}
	item, err := s.workqueue.GetWorkItemExecutionByEvent(ctx, eventID)
	if err != nil {
		return nil, err
	}
	if item == nil {
		return nil, ErrEventNotFound
	}
	if item.State != connector_workqueue.WorkStateDead {
		return nil, ErrEventNotDead
	}
	sanitizeDeadWorkItem(item)
	return item, nil
}

// GetDeadEvent preserves the published Event-shaped compatibility API while
// returning only the fixed-shape operator projection. Immutable Event payload
// and raw diagnostics remain internal execution provenance.
func (s *Service) GetDeadEvent(ctx context.Context, perms PermissionChecker, eventID string) (*ConnectorEvent, error) {
	item, err := s.GetDeadWorkItem(ctx, perms, eventID)
	if err != nil {
		return nil, err
	}
	ev := deadWorkItemToConnectorEvent(*item)
	return &ev, nil
}

// ListDeadWorkItems returns effective dead executor rows for a namespace.
//
// Requires perms.Has(constants.PermViewConnectorEvents); returns ErrPermissionDenied otherwise.
// Does not gate on pack activation state — once an event is enqueued, drain/finalize
// must work during pack removing/inactive transitions.
func (s *Service) ListDeadWorkItems(ctx context.Context, perms PermissionChecker, namespace string, page Page) ([]DeadWorkItem, error) {
	if !perms.Has(string(constants.PermViewConnectorEvents)) {
		return nil, ErrPermissionDenied
	}

	limit := page.Limit
	if limit <= 0 {
		limit = 50
	}
	offset := page.Offset
	if offset < 0 {
		offset = 0
	}
	items, err := s.workqueue.ListDeadWorkItems(ctx, connector_workqueue.ListDeadWorkItemsInput{
		Namespace: namespace,
		Limit:     limit,
		Offset:    offset,
	})
	if err != nil {
		return nil, err
	}
	for i := range items {
		sanitizeDeadWorkItem(&items[i])
	}
	return items, nil
}

// ListDeadEvents preserves the published Event-shaped compatibility API while
// delegating terminal state to the WorkItem-owned operator read path.
func (s *Service) ListDeadEvents(ctx context.Context, perms PermissionChecker, namespace string, page Page) ([]ConnectorEvent, error) {
	items, err := s.ListDeadWorkItems(ctx, perms, namespace, page)
	if err != nil {
		return nil, err
	}
	events := make([]ConnectorEvent, 0, len(items))
	for _, item := range items {
		events = append(events, deadWorkItemToConnectorEvent(item))
	}
	return events, nil
}

func deadWorkItemToConnectorEvent(item DeadWorkItem) ConnectorEvent {
	return ConnectorEvent{
		ID:               item.EventID,
		Namespace:        item.Namespace,
		InstalledPackID:  item.InstalledPackID,
		EventKind:        item.EventKind,
		IdempotencyKey:   item.IdempotencyKey,
		Status:           item.State,
		RetryCount:       item.FailureCount,
		LastErrorAt:      item.LastFailureAt,
		ProcessedAt:      item.ProcessedAt,
		ReplayedAt:       item.ReplayedAt,
		ReplayOfID:       item.ReplayOfEventID,
		DeliveryId:       item.DeliveryID,
		DeliverySequence: item.DeliverySequence,
		CreatedDate:      item.EventCreatedDate,
		LastModifiedDate: item.LastModifiedDate,
	}
}

// sanitizeDeadWorkItem keeps caller-authored failure diagnostics inside the
// executor repository. The permissioned operator service exposes fixed-shape
// state, counters, timestamps, and immutable provenance only.
func sanitizeDeadWorkItem(item *DeadWorkItem) {
	if item == nil {
		return
	}
	item.FailureClass = ""
	item.FailureSummary = ""
}

// ReplayDeadEvent inserts a new pending row that is a replay of the dead event
// identified by eventID.  The original dead row is preserved for audit with its
// ReplayedAt timestamp set atomically.
//
// packInstallID is the per-incarnation identity of the active connector pack
// the replay should be delivered to.  The caller (admin tool) decides which
// incarnation to replay against — typically the current active pack obtained
// via ResolveActivePack.  An empty packInstallID is an error.
//
// routeVersion is the RouteVersion to snapshot on the new replay row.
//
// # Caller contract (B14 fix / codex R11 BLOCK)
//
// The caller MUST resolve a current route via RouteResolver.ResolveFlow BEFORE
// calling ReplayDeadEvent, and pass the freshly resolved Version as routeVersion.
// Do NOT pass the original dead event's RouteVersion.
//
// Rationale: once an event reaches "dead" status, the route-incarnation and
// pack sweepers treat it as terminal and are free to GC the corresponding
// _System_ConnectorInboundRoute version row.  Passing the original dead event's
// RouteVersion can therefore result in ErrRouteVersionSwept / ErrRouteNotDeclared
// in the worker instead of a successful replay.  The caller re-resolves the
// current route; if the admin wants to replay with the original flow semantics,
// they should verify that RouteResolver.ResolveFlow returns the same
// FlowDeveloperName before proceeding. The service also preserves the durable
// source-projection versus per-event materialization class and rejects a route
// whose invocation mode would change that class.
//
// Accepted values:
//   - >= 1: specific route version resolved by RouteResolver.ResolveFlow.
//   - <= 0: invalid; returns ErrInvalidReplayRequest. The caller MUST resolve a
//     current route via RouteResolver.ResolveFlow (which returns >=1 or
//     ErrRouteNotDeclared, never 0) and pass the result here. The 0 sentinel is
//     the dispatcher's "no-route" marker — passing it to ReplayDeadEvent would
//     create a permanent-DLQ replay row that the worker immediately failTerminals
//     without ever processing, which is never the correct replay semantics.
//
// Concurrent-replay safety: the transaction follows active pack → Event →
// WorkItem, matching enqueue's pack-first order. For an execution generation
// that acquired WorkItem authority (Attempt > 0), the terminal WorkItem owns
// replay reservation. A pre-cut Event ReplayedAt marker remains a monotonic
// mixed-version "already replayed" fence, so an older service and this path
// cannot create two children. Other Event lifecycle fields are ignored for
// WorkItem-owned generations. Zero-attempt mirrors and Event-only generations
// remain on the explicit legacy Event reservation path. A concurrent loser
// returns ErrEventNotDead.
//
// A successful WorkItem-owned replay preserves the terminal original, inserts
// a new immutable Event, and creates a distinct WorkItem generation whose
// ReplayOfWorkId points to the original generation. Source-projection replay is
// rejected with ErrReplayBlocked when the target route changes execution class
// or another nonterminal generation exists in the immutable source scope.
//
// Atomicity: both reservations, the new Event, and the replay WorkItem run
// inside one transaction. Any pack, route, insert, or materialization failure
// rolls everything back so a future replay can retry safely.
//
// Requires perms.Has(constants.PermReplayConnectorEvents); returns ErrPermissionDenied otherwise.
// After the attested WorkItem-only cutover it returns
// [ErrLegacyEventReplayRetired] for a legacy Event-owned generation before its
// legacy replay reservation or a replay Event/WorkItem generation is created.
// WorkItem-owned dead generations retain their WorkItem replay path.
// Returns ErrEventNotDead when the event is not in dead state or replay already taken.
// Returns ErrReplayBlocked when source-projection class cannot be preserved or
// another generation is active in its source scope.
// Returns ErrInvalidReplayRequest when routeVersion <= 0.
func (s *Service) ReplayDeadEvent(ctx context.Context, perms PermissionChecker, eventID string, packInstallID string, routeVersion int64) (newEventID string, err error) {
	if !perms.Has(string(constants.PermReplayConnectorEvents)) {
		return "", ErrPermissionDenied
	}
	if strings.TrimSpace(packInstallID) == "" {
		return "", errors.New("connector-events: ReplayDeadEvent requires a non-empty packInstallID")
	}
	// B14 fix (extended): routeVersion must be >=1. The caller MUST resolve a
	// current route via RouteResolver.ResolveFlow (which returns >=1 or
	// ErrRouteNotDeclared, never 0) and pass the result here.
	//
	//   - Negative values: clearly invalid.
	//   - 0 (the dispatcher's "no-route" sentinel): also invalid as a replay
	//     argument. Passing 0 would create a new pending row that the worker
	//     immediately failTerminals without attempting to process, which is never
	//     correct replay semantics. If no route is declared, the caller should
	//     resolve the routing issue before replaying (fix the route config, then
	//     replay with the newly resolved version).
	if routeVersion <= 0 {
		return "", fmt.Errorf("%w: routeVersion must be >=1 (got %d; caller MUST resolve current route via RouteResolver.ResolveFlow first — passing 0 (no-route sentinel) or a negative value is a contract violation that would create a permanent-DLQ replay row)", ErrInvalidReplayRequest, routeVersion)
	}

	// Load immutable Event context. Effective terminal authority is resolved
	// inside the replay transaction.
	original, err := s.loadByID(ctx, eventID)
	if err != nil {
		return "", err
	}

	// B27 (codex R20): canonicalize the eventKind for the new replay row.
	// The original row's EventKind may be a pre-canonicalization raw value if it
	// was stored before the Enqueue normalization was added.  The replay INSERT
	// writes a new row, so its EventKind must match the canonical (lower-trimmed)
	// route row to avoid a JOIN miss / ErrPackIncarnationChanged on the next
	// Enqueue call for the same eventKind.  Uses the same form as load_route.go:36.
	replayEventKind := strings.ToLower(strings.TrimSpace(original.EventKind))

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

	// Replay creates a new Event and WorkItem generation. It is therefore an
	// ingress materializer, not an administrative bypass: lock the exact same
	// namespace fence in this transaction before the active-pack/Event/WorkItem
	// sequence below.
	if err = s.requireIngressAdmission(ctx, tx, original.Namespace); err != nil {
		return "", err
	}
	if s.afterIngressFenceLock != nil {
		s.afterIngressFenceLock()
	}

	// Enqueue owns the active-pack row before inserting an Event fact or
	// reconciling its WorkItem. Replay follows the same global order:
	//
	//	active pack -> Event -> WorkItem
	//
	// Completion and retry do not need the pack row. This prevents
	// replay/enqueue wait cycles.
	if err = lockActivePackIncarnation(ctx, tx, original.Namespace, packInstallID); err != nil {
		return "", err
	}
	cutoverAttested, err := rejectLegacyEventReplayAfterWorkItemOnlyCutover(ctx, tx, eventID)
	if err != nil {
		return "", err
	}
	// Lock immutable Event context before its WorkItem.
	// Mutable Event lifecycle fields are consulted only when no WorkItem-owned
	// generation exists and replay must remain on the legacy path.
	now := s.now()
	var eventStatus string
	var eventReplayedAt sql.NullTime
	if s.beforeReplayEventLock != nil {
		s.beforeReplayEventLock()
	}
	if err = tx.QueryRowContext(ctx, `
SELECT Status, ReplayedAt
  FROM _System_ConnectorEvent
 WHERE Id = ?
 FOR UPDATE`, eventID).Scan(&eventStatus, &eventReplayedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return "", ErrEventNotFound
		}
		return "", err
	}
	if s.afterReplayEventLock != nil {
		s.afterReplayEventLock()
	}
	// Older binaries reserve replay on Event.ReplayedAt before they know about
	// WorkItem replay authority. Treat that committed marker as a fail-closed
	// compatibility fence during rolling upgrade. The WorkItem-owned path never
	// writes it, and R1.4 can retire this read after the compatibility window.
	if eventReplayedAt.Valid {
		return "", ErrEventNotDead
	}
	var replayReservation connector_workqueue.ReplayReservation
	reserveWorkItem := func() error {
		var reserveErr error
		replayReservation, reserveErr = s.workqueue.WithTx(tx).ReserveDeadWorkItemReplay(ctx, eventID, workTypeForEventKind(original.EventKind), now)
		if errors.Is(reserveErr, connector_workqueue.ErrReplayBlocked) {
			return ErrReplayBlocked
		}
		return reserveErr
	}

	if err = reserveWorkItem(); err != nil {
		return "", err
	}
	if cutoverAttested && !replayReservation.WorkItemOwned {
		return "", ErrLegacyEventReplayRetired
	}
	if replayReservation.WorkItemOwned {
		if !replayReservation.Reserved {
			return "", ErrEventNotDead
		}
	} else if eventStatus == statusDead {
		res, execErr := tx.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
   SET ReplayedAt = ?,
       LastModifiedDate = ?
 WHERE Id = ?
   AND Status = ?
   AND ReplayedAt IS NULL`,
			now, now, eventID, statusDead)
		if execErr != nil {
			return "", execErr
		}
		reserved, rowsErr := res.RowsAffected()
		if rowsErr != nil {
			return "", rowsErr
		}
		if reserved == 0 {
			return "", ErrEventNotDead
		}
	} else {
		if !replayRepairableEventStatus(eventStatus) {
			return "", ErrEventNotDead
		}
		return "", ErrEventNotDead
	}
	if !replayReservation.WorkItemOwned && replayReservation.WorkID != "" {
		fenced, fenceErr := s.workqueue.WithTx(tx).FenceZeroAttemptReplayMirror(ctx, eventID, replayReservation.WorkID, now)
		if fenceErr != nil {
			return "", fenceErr
		}
		if !fenced {
			return "", ErrReplayBlocked
		}
	}
	replaySourceProjection, replayClassKnown, classErr := s.resolveReplayExecutionClass(
		ctx, tx, original, replayReservation,
	)
	if classErr != nil {
		return "", classErr
	}
	if !replayClassKnown {
		return "", ErrReplayBlocked
	}
	replayEnqueueOrdinal, err := s.nextEnqueueOrdinal(ctx, tx, packInstallID)
	if err != nil {
		return "", err
	}

	// Insert a new replay row inside the same txn — atomic with active-pack guard AND
	// route-liveness guard (R22, symmetric to Enqueue's R17-R21 fix).
	//
	// The INSERT…SELECT checks two things at INSERT-time:
	//   1. Active-pack guard: (Name=namespace, PackInstallId=packInstallID, Status='active')
	//      still holds.  Pack rotated? RowsAffected=0 → ErrPackIncarnationChanged.
	//   2. Route-liveness guard: the caller-resolved routeVersion still exists in
	//      _System_ConnectorInboundRoute with IsDeleted=0, IsCurrent=1.  Admin may have
	//      resolved a current route version, and then the pack manifest may have dropped
	//      that event kind before the replay INSERT; the sweeper either clears IsCurrent
	//      (preserved-for-backlog) or soft-deletes (no non-terminal references remain).
	//      If the route has been swept, RowsAffected=0 → ErrPackIncarnationChanged, and
	//      the transaction rollback restores ReplayedAt=NULL so the admin can retry after
	//      re-resolving the route.
	//
	// Both guards are enforced for routeVersion >= 1 (the only value callers may pass;
	// see the validation block above — routeVersion <= 0 is rejected before this point).
	newID := s.idgen()
	// Generate a fresh idempotency key so the unique constraint is not violated.
	// Use "replay:"+newID (7+36=43 chars) rather than embedding the original key,
	// which could overflow VARCHAR(255) for long original keys. The linkage to the
	// original event is already captured by ReplayOfId.
	replayKey := "replay:" + newID
	var replayResult sql.Result
	var replayCompanionPackArg interface{}
	var replayCompanionFlowArg interface{}
	var replaySourceObjectKindArg interface{}
	var replaySourceObjectIDArg interface{}
	var replaySourceLaneKeyArg interface{}
	replayLane := sourceLaneFromPayload(original.Payload)
	if original.CompanionPackInstallID != "" {
		replayCompanionPackArg = original.CompanionPackInstallID
	}
	if original.CompanionFlowVersionID != "" {
		replayCompanionFlowArg = original.CompanionFlowVersionID
	}
	if replayLane.ObjectKind != "" {
		replaySourceObjectKindArg = replayLane.ObjectKind
	}
	if replayLane.SourceID != "" {
		replaySourceObjectIDArg = replayLane.SourceID
	}
	if replayLane.Key != "" {
		replaySourceLaneKeyArg = replayLane.Key
	}
	if storedLaneKey := strings.TrimSpace(original.SourceLaneKey); storedLaneKey != "" {
		replaySourceLaneKeyArg = storedLaneKey
	}
	// Replay preserves the enqueue-time companion snapshot: CompanionPackInstallId
	// and CompanionFlowVersionId are copied verbatim from the dead row. Correctness
	// across companion evolution is enforced at drain by the worker's incarnation
	// re-check (§5.2) — if the frozen companion incarnation is no longer installed,
	// the worker fails the event closed rather than executing a stale/wrong version.
	// We deliberately do NOT add an enqueue-time companion guard on the replay path:
	// the provider active-pack + route-liveness guards below cover provider rotation;
	// re-targeting a dead cross-pack event at a freshly-evolved companion incarnation
	// is intentionally out of scope (the snapshot is incarnation-bound — fresh-snapshot
	// replay recovery is future work, not part of this contract). Guarding replay on
	// companion-active would instead make such events permanently unreplayable.
	replayQuery := `
INSERT INTO _System_ConnectorEvent
  (Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, QueueClass, Payload, RetryCount,
   SourceObjectKind, SourceObjectID, SourceLaneKey, RouteVersion, CompanionPackInstallId, CompanionFlowVersionId, ReplayOfId, DeliveryId, DeliverySequence, EnqueueOrdinal, CreatedDate, LastModifiedDate)
SELECT ?, ?, ?, ?, ?, ?, COALESCE(NULLIF(r.QueueClass, ''), 'default'), ?, 0, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
  FROM _System_ConnectorInboundRoute r
 WHERE r.InstalledPackId = ?
   AND r.EventKind       = ?
   AND r.Version         = ?
   AND COALESCE(r.IsDeleted, 0) = 0
   AND COALESCE(r.IsCurrent, 0) = 1`
	// Replay rejoins the original delivery's ordering group (issue #2154) ONLY within
	// the SAME pack incarnation: copy DeliveryId + DeliverySequence from the dead row so
	// the replayed event cannot execute out-of-order relative to still-pending siblings
	// of that delivery. A CROSS-incarnation replay (the dead row was enqueued under a
	// superseded PackInstallId, and the caller resolved the current active incarnation,
	// so original.InstalledPackID != packInstallID) does NOT copy them: the original
	// delivery's other members live under a different InstalledPackId that ClaimNext —
	// which is incarnation-scoped — cannot co-order with, and those superseded-incarnation
	// events are stranded by the rebirth lifecycle (ListConnectorPacksForInboundWorkers
	// drains only one incarnation per namespace). Copying the DeliveryId there would
	// falsely advertise a group this row can never serialize against. Such a replay, and
	// a legacy dead row with an empty DeliveryId, replay as an independent NULL singleton;
	// #2134's IsRetriable retry is the ordering backstop for those residual cases.
	var replayDeliveryIDArg interface{}
	replayDeliverySeq := 0
	if strings.TrimSpace(original.DeliveryId) != "" &&
		strings.TrimSpace(original.InstalledPackID) == strings.TrimSpace(packInstallID) {
		replayDeliveryIDArg = strings.TrimSpace(original.DeliveryId)
		replayDeliverySeq = original.DeliverySequence
	}
	replayArgs := []interface{}{
		newID, original.Namespace, packInstallID, replayEventKind, replayKey,
		statusPending, nullableJSON(original.Payload), replaySourceObjectKindArg, replaySourceObjectIDArg, replaySourceLaneKeyArg, routeVersion,
		replayCompanionPackArg, replayCompanionFlowArg, eventID,
		replayDeliveryIDArg, replayDeliverySeq, replayEnqueueOrdinal, now, now,
		packInstallID, replayEventKind, routeVersion,
	}
	replayQuery += "\nLIMIT 1"
	replayResult, err = tx.ExecContext(ctx, replayQuery, replayArgs...)
	if err != nil {
		return "", err
	}
	var rn int64
	rn, err = replayResult.RowsAffected()
	if err != nil {
		return "", err
	}
	if rn == 0 {
		// Pack rotated between caller's resolve and this INSERT — txn will rollback,
		// restoring ReplayedAt=NULL so a future ReplayDeadEvent call can succeed.
		err = ErrPackIncarnationChanged
		return "", err
	}
	// A zero-attempt compatibility mirror does not own replay reservation, but
	// its WorkID still supplies immutable lineage. Event-only generations leave
	// lineage empty. Both paths retain explicit replay semantics so a derived
	// source ProjectionKey cannot retarget an active successor.
	replayOfWorkID := replayReservation.WorkID
	materialized, materializeErr := s.ensureWorkItemForEventWithReplay(
		ctx, tx, newID, "", now, false, true, replayOfWorkID, true, original.Namespace,
		replaySourceProjection,
	)
	if errors.Is(materializeErr, connector_workqueue.ErrReplayBlocked) {
		return "", ErrReplayBlocked
	}
	if materializeErr != nil {
		return "", materializeErr
	}
	if replayReservation.WorkID != "" && materialized == nil {
		return "", errors.New("connector-events: WorkItem lineage replay did not materialize a replay WorkItem")
	}

	if err = tx.Commit(); err != nil {
		return "", err
	}
	return newID, nil
}

func replayRepairableEventStatus(status string) bool {
	switch strings.TrimSpace(status) {
	case statusPending, statusProcessing, statusFailed:
		return true
	default:
		return false
	}
}

// resolveReplayExecutionClass derives the original generation's durable
// materialization class. WorkItem generation identity is authoritative when it
// exists. Event-only compatibility rows use their immutable route snapshot; a
// missing snapshot fails closed when a source lane could have been projected.
func (s *Service) resolveReplayExecutionClass(
	ctx context.Context,
	tx *sql.Tx,
	original *ConnectorEvent,
	reservation connector_workqueue.ReplayReservation,
) (sourceProjection bool, known bool, err error) {
	if original == nil {
		return false, false, ErrEventNotFound
	}
	if strings.TrimSpace(reservation.WorkID) != "" {
		return connector_workqueue.IsSourceProjectionWorkID(reservation.WorkID) ||
			strings.TrimSpace(reservation.ProjectionKey) != "", true, nil
	}
	if strings.TrimSpace(original.SourceLaneKey) == "" {
		return false, true, nil
	}
	if !original.RouteVersion.Valid || original.RouteVersion.Int64 <= 0 {
		return false, false, nil
	}

	var invocationMode string
	err = tx.QueryRowContext(ctx, `
SELECT COALESCE(InvocationMode, '')
  FROM _System_ConnectorInboundRoute
 WHERE InstalledPackId = ?
   AND LOWER(TRIM(EventKind)) = ?
   AND Version = ?
 LIMIT 1`,
		strings.TrimSpace(original.InstalledPackID),
		strings.ToLower(strings.TrimSpace(original.EventKind)),
		original.RouteVersion.Int64,
	).Scan(&invocationMode)
	if errors.Is(err, sql.ErrNoRows) {
		return false, false, nil
	}
	if err != nil {
		return false, false, err
	}
	return strings.EqualFold(
		strings.TrimSpace(invocationMode),
		connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled,
	), true, nil
}