package persistence

import (
	"context"
	"fmt"
	"sync"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/kernel/pkg/ports"
)

const DefaultParentLaneHeartbeatInterval = DefaultParentLaneLeaseDuration / 3

// ParentLaneHeartbeatOptions controls transaction-lifetime parent-lane
// heartbeat intervals and cleanup timeouts.
type ParentLaneHeartbeatOptions struct {
	Interval       time.Duration
	Lease          time.Duration
	ReleaseTimeout time.Duration

	observer parentLaneHeartbeatObserver
}

// ParentLaneHeartbeatController owns the heartbeat loop for one transaction's
// acquired parent lanes.
type ParentLaneHeartbeatController struct {
	store *ParentLaneStore

	txCtx    context.Context
	ctx      context.Context
	cancel   context.CancelFunc
	done     chan struct{}
	stopOnce sync.Once

	mu       sync.Mutex
	owners   []ParentLaneOwnership
	stopping bool

	interval       time.Duration
	lease          time.Duration
	releaseTimeout time.Duration
	observer       parentLaneHeartbeatObserver
}

type parentLaneHeartbeatObserver func(context.Context, parentLaneHeartbeatEvent)

type parentLaneHeartbeatEvent struct {
	action    string
	result    string
	laneCount int
	keyHash   string
	outcome   ports.TransactionCompletionOutcome
	err       error
}

// StartHeartbeat begins transaction-lifetime heartbeat management for acquired
// parent-lane owners. It performs one token-checked heartbeat before returning,
// so callers have live ownership evidence before entering business locks.
// Heartbeat execution failure and known lost ownership are fail-closed as
// retryable parent-lane contention: the current transaction is cancelled/aborted
// and the PR2B0 pre-commit veto prevents a later unsafe commit. Release is
// token-checked for every owner after any transaction completion outcome.
func (s *ParentLaneStore) StartHeartbeat(ctx context.Context, owners []ParentLaneOwnership, opts ParentLaneHeartbeatOptions) (*ParentLaneHeartbeatController, error) {
	controller, err := newParentLaneHeartbeatController(ctx, s, owners, opts)
	if err != nil || len(controller.owners) == 0 {
		return controller, err
	}
	if ok, err := controller.heartbeatOnceErr(ctx); !ok {
		return nil, err
	}
	if !ports.RegisterTransactionCompletionHook(ctx, controller.complete) {
		return nil, apperrors.NewInternalError("parent lane heartbeat requires transaction completion hooks", nil)
	}
	go controller.run()
	return controller, nil
}

func newParentLaneHeartbeatController(ctx context.Context, store *ParentLaneStore, owners []ParentLaneOwnership, opts ParentLaneHeartbeatOptions) (*ParentLaneHeartbeatController, error) {
	if store == nil || store.backend == nil {
		return nil, apperrors.NewInternalError("parent lane store is not configured", nil)
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if !ports.TransactionAbortSupported(ctx) && len(owners) > 0 {
		return nil, apperrors.NewInternalError("parent lane heartbeat requires transaction abort support", nil)
	}
	copied := make([]ParentLaneOwnership, len(owners))
	copy(copied, owners)
	for _, owner := range copied {
		if owner.KeyHash == "" || owner.OwnerToken == "" {
			return nil, apperrors.NewInternalError("parent lane heartbeat requires acquired owner token and key hash", nil)
		}
	}

	lease := opts.Lease
	if lease <= 0 {
		lease = store.lease
	}
	if lease <= 0 {
		lease = DefaultParentLaneLeaseDuration
	}
	interval := opts.Interval
	if interval <= 0 {
		interval = lease / 3
	}
	if interval <= 0 {
		interval = DefaultParentLaneHeartbeatInterval
	}
	if interval <= 0 {
		interval = time.Millisecond
	}
	releaseTimeout := opts.ReleaseTimeout
	if releaseTimeout <= 0 {
		releaseTimeout = store.releaseTimeout
	}
	if releaseTimeout <= 0 {
		releaseTimeout = DefaultParentLaneCompletionReleaseTimeout
	}
	observer := opts.observer
	if observer == nil {
		observer = observeParentLaneHeartbeat
	}
	loopCtx, cancel := context.WithCancel(ctx)
	controller := &ParentLaneHeartbeatController{
		store:          store,
		txCtx:          ctx,
		ctx:            loopCtx,
		cancel:         cancel,
		done:           make(chan struct{}),
		owners:         copied,
		interval:       interval,
		lease:          lease,
		releaseTimeout: releaseTimeout,
		observer:       observer,
	}
	if len(copied) == 0 {
		close(controller.done)
	}
	return controller, nil
}

// Done is closed after the heartbeat loop exits.
func (c *ParentLaneHeartbeatController) Done() <-chan struct{} {
	if c == nil {
		done := make(chan struct{})
		close(done)
		return done
	}
	return c.done
}

// AddOwner registers a newly acquired lane with the running transaction heartbeat.
func (c *ParentLaneHeartbeatController) AddOwner(owner ParentLaneOwnership) error {
	if c == nil {
		return apperrors.NewInternalError("parent lane heartbeat controller is not configured", nil)
	}
	if owner.KeyHash == "" || owner.OwnerToken == "" {
		return apperrors.NewInternalError("parent lane heartbeat requires acquired owner token and key hash", nil)
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.stopping || c.ctx.Err() != nil {
		return apperrors.NewLockContentionError(ErrParentLaneContention)
	}
	c.owners = append(c.owners, owner)
	return nil
}

func (c *ParentLaneHeartbeatController) run() {
	defer close(c.done)
	for {
		if err := c.store.clock.Sleep(c.ctx, c.interval); err != nil {
			if !c.isStopping() {
				c.failClosed(parentLaneHeartbeatEvent{action: "stop", result: "context_cancelled", laneCount: c.ownerCount(), err: err})
			}
			return
		}
		if !c.heartbeatOnce(c.ctx) {
			return
		}
	}
}

func (c *ParentLaneHeartbeatController) heartbeatOnce(ctx context.Context) bool {
	ok, _ := c.heartbeatOnceErr(ctx)
	return ok
}

func (c *ParentLaneHeartbeatController) heartbeatOnceErr(ctx context.Context) (bool, error) {
	owners := c.snapshotOwners()
	if len(owners) == 0 {
		return true, nil
	}
	for i, owner := range owners {
		if c.isStopping() {
			return false, nil
		}
		next, ok, err := c.store.Heartbeat(ctx, owner, c.lease)
		if err != nil {
			// Transaction completion marks the controller as stopping before it
			// cancels the heartbeat context. An in-flight SQL heartbeat therefore
			// returns context.Canceled during normal completion; completion owns
			// the token-checked release and this is not a fail-closed condition.
			if c.isStopping() {
				return false, nil
			}
			event := parentLaneHeartbeatEvent{action: "extend", result: "failure", laneCount: 1, keyHash: shortParentLaneHash(owner.KeyHash), err: err}
			c.observe(event)
			return false, c.failClosed(event)
		}
		if !ok {
			if c.isStopping() {
				return false, nil
			}
			event := parentLaneHeartbeatEvent{action: "extend", result: "lost_owner", laneCount: 1, keyHash: shortParentLaneHash(owner.KeyHash)}
			c.observe(event)
			return false, c.failClosed(event)
		}
		c.mu.Lock()
		if i < len(c.owners) && c.owners[i].KeyHash == owner.KeyHash && c.owners[i].OwnerToken == owner.OwnerToken {
			c.owners[i] = next
		}
		c.mu.Unlock()
	}
	c.observe(parentLaneHeartbeatEvent{action: "extend", result: "success", laneCount: len(owners)})
	return true, nil
}

func (c *ParentLaneHeartbeatController) complete(outcome ports.TransactionCompletionOutcome) {
	c.stop()
	<-c.done
	c.releaseAll(outcome)
}

func (c *ParentLaneHeartbeatController) stop() {
	c.stopOnce.Do(func() {
		c.mu.Lock()
		c.stopping = true
		c.mu.Unlock()
		c.cancel()
	})
}

func (c *ParentLaneHeartbeatController) isStopping() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.stopping
}

func (c *ParentLaneHeartbeatController) ownerCount() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return len(c.owners)
}

func (c *ParentLaneHeartbeatController) snapshotOwners() []ParentLaneOwnership {
	c.mu.Lock()
	defer c.mu.Unlock()
	owners := make([]ParentLaneOwnership, len(c.owners))
	copy(owners, c.owners)
	return owners
}

func (c *ParentLaneHeartbeatController) failClosed(event parentLaneHeartbeatEvent) error {
	cause := parentLaneHeartbeatFailClosedCause(event)
	if !ports.AbortTransaction(c.txCtx, cause) {
		event.action = "abort"
		event.result = "unsupported"
		event.err = cause
		c.observe(event)
	}
	c.cancel()
	return cause
}

func parentLaneHeartbeatFailClosedCause(event parentLaneHeartbeatEvent) error {
	if event.action == "stop" && event.err != nil {
		return event.err
	}
	if event.err == nil {
		return apperrors.NewLockContentionError(ErrParentLaneContention)
	}
	return apperrors.NewLockContentionError(fmt.Errorf("%w: heartbeat failed: %v", ErrParentLaneContention, event.err))
}

func (c *ParentLaneHeartbeatController) releaseAll(outcome ports.TransactionCompletionOutcome) {
	owners := c.snapshotOwners()
	released := 0
	for _, owner := range owners {
		ok, err := c.store.releaseWithCompletionBudget(owner, c.releaseTimeout)
		if err != nil {
			c.observe(parentLaneHeartbeatEvent{action: "release", result: "failure", laneCount: 1, keyHash: shortParentLaneHash(owner.KeyHash), outcome: outcome, err: err})
			continue
		}
		if !ok {
			c.observe(parentLaneHeartbeatEvent{action: "release", result: "stale_owner", laneCount: 1, keyHash: shortParentLaneHash(owner.KeyHash), outcome: outcome})
			continue
		}
		released++
	}
	if released > 0 {
		c.observe(parentLaneHeartbeatEvent{action: "release", result: "success", laneCount: released, outcome: outcome})
	}
}

func (c *ParentLaneHeartbeatController) observe(event parentLaneHeartbeatEvent) {
	if c == nil || c.observer == nil {
		return
	}
	c.observer(c.txCtx, event)
}

func observeParentLaneHeartbeat(ctx context.Context, event parentLaneHeartbeatEvent) {
	if event.result == "success" && !perftiming.Enabled(ctx) {
		return
	}
	logEvent := perftiming.BaseEvent(ctx, "parent_lane_heartbeat", event.result, event.err).
		Str("action", event.action).
		Int("lane_count", event.laneCount)
	if event.keyHash != "" {
		logEvent = logEvent.Str("key_hash", event.keyHash)
	}
	if event.outcome != "" {
		logEvent = logEvent.Str("outcome", string(event.outcome))
	}
	logEvent.Msg("Parent lane heartbeat lifecycle")
}

func shortParentLaneHash(hash string) string {
	if len(hash) <= 12 {
		return hash
	}
	return hash[:12]
}