package connector_inbound

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"time"

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

	zlog "github.com/rs/zerolog/log"
)

// errFlowDeadlineExceeded is the cause passed to ReleaseClaimForRetry when the
// opt-in FlowDeadline fires. It is written to LastError so operators can tell a
// deadline re-drive apart from a flow-signalled retriable defer.
var errFlowDeadlineExceeded = errors.New("connector_inbound: flow execution exceeded FlowDeadline; released for retry")

// errBatchReadDeadlineExceeded is the cause passed to ReleaseClaimForRetry when a
// post-claim BatchRead exceeds FlowDeadline. Like a hung flow, a wedged read is
// hung-but-not-dead — released for retry, never MarkFailed (no dead-letter burn).
var errBatchReadDeadlineExceeded = errors.New("connector_inbound: post-claim batch read exceeded FlowDeadline; released for retry")

// leaseRenewingEventsService is the optional capability the worker uses to renew
// the lease of an in-flight claimed event (RenewLease bumps LastModifiedDate),
// keeping a slow-but-alive flow's row past the StaleProcessingSeconds reclaim
// cutoff so stale-reclaim cannot double-execute it. The production
// connector_events.Service implements it; if the injected service does not, the
// worker runs without a heartbeat (fail-soft, like staleReclaimingEventsService
// and claimReleasingEventsService) — it never panics on the missing capability.
type leaseRenewingEventsService interface {
	RenewLease(ctx context.Context, eventID, claimToken string) error
}

// leaseRef identifies one lease the worker holds: the event id and the per-claim
// token minted by ClaimNext/ClaimByID for this grant. The heartbeat renews each
// ref under its own token so a batch group's N members all stay fresh.
type leaseRef struct {
	eventID    string
	claimToken string
}

// leaseRefsFor returns the lease refs covering an in-flight unit. For a single
// event it is one ref; the batch path passes the whole group.
func leaseRefsFor(ev *connector_events.ConnectorEvent) []leaseRef {
	if ev == nil || ev.ClaimToken == "" {
		return nil
	}
	return []leaseRef{{eventID: ev.ID, claimToken: ev.ClaimToken}}
}

// leaseRefsForGroup returns one ref per group member (each carries its own token).
func leaseRefsForGroup(group *pendingBatchGroup) []leaseRef {
	if group == nil {
		return nil
	}
	refs := make([]leaseRef, 0, len(group.events))
	for _, item := range group.events {
		if item.ev != nil && item.ev.ClaimToken != "" {
			refs = append(refs, leaseRef{eventID: item.ev.ID, claimToken: item.ev.ClaimToken})
		}
	}
	return refs
}

// leaseHeartbeatInterval returns the renewal period: the configured value, or
// StaleProcessingSeconds/3 (~100 s for the 300 s cutoff) by default. Dividing by
// 3 gives two renewals of headroom before the cutoff even if one tick is delayed.
func (w *Worker) leaseHeartbeatInterval() time.Duration {
	if w.cfg.LeaseHeartbeatInterval > 0 {
		return w.cfg.LeaseHeartbeatInterval
	}
	return connector_events.StaleProcessingSeconds * time.Second / 3
}

// startLeaseHeartbeat starts a single goroutine that periodically renews every
// lease in refs until the returned stop func is called. The stop func is
// idempotent and BLOCKS until the goroutine has exited, so by the time the caller
// finalizes the in-flight unit no further renewal can fire (deterministic
// lifecycle: no leaked goroutine, no post-finalize renewal racing a reclaim).
//
// Renewals use a fresh background context (not the flow ctx): a flow ctx that is
// cancelled by FlowDeadline must NOT also kill the heartbeat write — the renewal
// is a tiny point UPDATE and is naturally a no-op once the row finalizes
// (RenewLease returns ErrLeaseLost, which is swallowed here).
//
// Returns a no-op stop when the service lacks RenewLease or there is nothing to
// renew, so callers can always defer stop() unconditionally.
func (w *Worker) startLeaseHeartbeat(refs []leaseRef) (stop func()) {
	renewer, ok := w.events.(leaseRenewingEventsService)
	if !ok || len(refs) == 0 {
		return func() {}
	}
	interval := w.leaseHeartbeatInterval()
	if interval <= 0 {
		return func() {}
	}

	done := make(chan struct{})
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-done:
				return
			case <-ticker.C:
				w.renewAll(renewer, refs)
			}
		}
	}()

	var once sync.Once
	return func() {
		once.Do(func() {
			close(done)
			wg.Wait()
		})
	}
}

// renewAll renews every ref once. A renewal that finds the lease gone
// (ErrLeaseLost) is benign — the row finalized or was reclaimed — and is logged
// at debug only; any other error is logged at warn but never propagated (a
// heartbeat failure must not fail the event).
func (w *Worker) renewAll(renewer leaseRenewingEventsService, refs []leaseRef) {
	for _, ref := range refs {
		ctx, cancel := w.finalizeContext()
		err := renewer.RenewLease(ctx, ref.eventID, ref.claimToken)
		cancel()
		if err == nil || errors.Is(err, connector_events.ErrLeaseLost) {
			continue
		}
		zlog.Warn().
			Err(err).
			Str("worker_id", w.cfg.WorkerID).
			Str("event_id", ref.eventID).
			Msg("connector inbound worker: lease heartbeat renewal failed (continuing)")
	}
}

// batchReadHeartbeated runs a post-claim batchRead under BOTH a lease heartbeat
// and the flow deadline — the same alive/wedged pairing the flow span uses.
//
//   - Heartbeat (for the slow-but-ALIVE read): renews each held lease for the
//     read's duration, so a legitimately slow retry-backed BatchRead does not go
//     stale mid-read and get reclaimed + double-executed. The stop is DEFERRED so
//     a recovered BatchRead panic (a supported path — see worker_panic_recovery_test)
//     unwinds through it and never leaks the renewer goroutine.
//   - Deadline (for the WEDGED read): the heartbeat would otherwise renew a hung
//     vendor read's lease forever and pin its lane, since stale-reclaim can no
//     longer recover a heartbeated row. flowDeadlineContext bounds the read at the
//     same generous FlowDeadline the flow span uses; a wedged read's ctx then
//     cancels and the read errors. The returned bool reports whether THIS read hit
//     the deadline, so the caller re-drives via ReleaseClaimForRetry (NOT
//     MarkFailed — a hung-but-not-dead read must not burn the dead-letter budget),
//     exactly like the flow-deadline path. With FlowDeadline disabled (env 0) the
//     read is unbounded, exactly as the flow span is.
//
// The flow span is heartbeated + deadline-bounded separately inside
// executeEvent/executeBatchCollection.
func (w *Worker) batchReadHeartbeated(ctx context.Context, refs []leaseRef, objectKind string, sourceIDs []string) (map[string]json.RawMessage, time.Duration, bool, error) {
	stop := w.startLeaseHeartbeat(refs)
	defer stop()
	readCtx, cancel := w.flowDeadlineContext(ctx)
	defer cancel()
	details, elapsed, err := w.batchReadWithStageTiming(readCtx, objectKind, sourceIDs)
	return details, elapsed, isFlowDeadlineExceeded(readCtx, err), err
}

func (w *Worker) batchReadOutcomeHeartbeated(ctx context.Context, refs []leaseRef, objectKind string, sourceIDs []string) (BatchReadOutcome, time.Duration, bool, error) {
	stop := w.startLeaseHeartbeat(refs)
	defer stop()
	readCtx, cancel := w.flowDeadlineContext(ctx)
	defer cancel()
	outcome, elapsed, err := w.batchReadOutcomeWithStageTiming(readCtx, objectKind, sourceIDs)
	deadlineHit := isFlowDeadlineExceeded(readCtx, err)
	if !deadlineHit {
		for _, sourceErr := range outcome.SourceErrors {
			if isFlowDeadlineExceeded(readCtx, sourceErr) {
				deadlineHit = true
				break
			}
		}
	}
	return outcome, elapsed, deadlineHit, err
}

// flowDeadlineContext wraps ctx in a timeout when FlowDeadline > 0, bounding a
// post-claim external call (flow execution, or a heartbeated post-claim BatchRead)
// that would otherwise heartbeat forever and pin its lane. With FlowDeadline == 0
// it returns ctx unchanged with a no-op cancel, so a legitimately slow-but-
// completing call is never interrupted.
func (w *Worker) flowDeadlineContext(ctx context.Context) (context.Context, context.CancelFunc) {
	if w.cfg.FlowDeadline <= 0 {
		return ctx, func() {}
	}
	return context.WithTimeout(ctx, w.cfg.FlowDeadline)
}

// isFlowDeadlineExceeded reports whether err is the FlowDeadline timeout firing
// (the flow ctx hit context.DeadlineExceeded). Used to route a deadline-exceeded
// flow through ReleaseClaimForRetry rather than MarkFailed.
func isFlowDeadlineExceeded(ctx context.Context, err error) bool {
	return err != nil && errors.Is(ctx.Err(), context.DeadlineExceeded)
}