package records

import (
	"context"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/persistence"
)

// ParentLaneHeartbeatController is the records-layer view of a transaction
// heartbeat controller. Production uses persistence.ParentLaneHeartbeatController;
// tests can inject a fake without a SQL backend.
type ParentLaneHeartbeatController interface {
	AddOwner(persistence.ParentLaneOwnership) error
	Done() <-chan struct{}
}

// ParentLaneAdmissionController is the narrow records-layer dependency for
// durable parent-lane admission. It owns acquire, transaction-lifetime heartbeat,
// and best-effort cleanup when admission fails before the heartbeat is armed.
type ParentLaneAdmissionController interface {
	Acquire(context.Context, recordlocks.ParentKey, persistence.ParentLaneAcquireOptions) (persistence.ParentLaneOwnership, error)
	StartHeartbeat(context.Context, []persistence.ParentLaneOwnership, persistence.ParentLaneHeartbeatOptions) (ParentLaneHeartbeatController, error)
	Release(context.Context, persistence.ParentLaneOwnership) (bool, error)
}

type parentLaneAdmissionCleanupEvent struct {
	result    string
	laneCount int
	keyHash   string
	err       error
}

type parentLaneAdmissionCleanupObserver func(context.Context, parentLaneAdmissionCleanupEvent)

var parentLaneAdmissionCleanupReleaseTimeout = persistence.DefaultParentLaneReleaseTimeout
var observeParentLaneAdmissionCleanup parentLaneAdmissionCleanupObserver = logParentLaneAdmissionCleanup

type persistenceParentLaneAdmissionController struct {
	store *persistence.ParentLaneStore
}

// NewParentLaneAdmissionController adapts the production persistence store to
// the narrow records-layer controller seam.
func NewParentLaneAdmissionController(store *persistence.ParentLaneStore) ParentLaneAdmissionController {
	if store == nil {
		return nil
	}
	return persistenceParentLaneAdmissionController{store: store}
}

func (c persistenceParentLaneAdmissionController) Acquire(ctx context.Context, key recordlocks.ParentKey, opts persistence.ParentLaneAcquireOptions) (persistence.ParentLaneOwnership, error) {
	return c.store.Acquire(ctx, key, opts)
}

func (c persistenceParentLaneAdmissionController) StartHeartbeat(ctx context.Context, owners []persistence.ParentLaneOwnership, opts persistence.ParentLaneHeartbeatOptions) (ParentLaneHeartbeatController, error) {
	return c.store.StartHeartbeat(ctx, owners, opts)
}

func (c persistenceParentLaneAdmissionController) Release(ctx context.Context, owner persistence.ParentLaneOwnership) (bool, error) {
	return c.store.Release(ctx, owner)
}

// SetParentLaneAdmissionController injects the parent-lane admission dependency.
// Production wiring uses NewParentLaneAdmissionController(persistence.NewParentLaneStore(conn));
// records tests can inject a fake through the same seam.
func (ps *PersistenceService) SetParentLaneAdmissionController(controller ParentLaneAdmissionController) {
	ps.parentLaneAdmission = controller
}

// SetParentLaneStore is a convenience production-style setter mirroring SetRawDB.
func (ps *PersistenceService) SetParentLaneStore(store *persistence.ParentLaneStore) {
	ps.SetParentLaneAdmissionController(NewParentLaneAdmissionController(store))
}

// HasParentLaneAdmissionController lets composition-root tests assert the
// production dependency is present before a future admitted hot path can run.
func (ps *PersistenceService) HasParentLaneAdmissionController() bool {
	return ps != nil && ps.parentLaneAdmission != nil
}

// RunWithParentLaneAdmission is the fail-closed seam for admitted parent-row
// lock paths. Call it inside the pessimistic transaction before business
// parent-row FOR UPDATE locks.
func (ps *PersistenceService) RunWithParentLaneAdmission(ctx context.Context, keys []recordlocks.ParentKey, fn func(context.Context) error) error {
	if fn == nil {
		return nil
	}
	ordered := recordlocks.OrderedParentKeys(keys)
	if len(ordered) == 0 {
		return fn(ctx)
	}
	if ps == nil || ps.parentLaneAdmission == nil {
		return appErrors.NewInternalError("parent lane admission requires a configured controller", nil)
	}
	if ps.txManager == nil || !ps.txManager.IsTransactionActive(ctx) || !ps.txManager.IsPessimisticTx(ctx) {
		return appErrors.NewInternalError("parent lane admission requires an active pessimistic transaction", nil)
	}
	if !ports.TransactionAbortSupported(ctx) {
		return appErrors.NewInternalError("parent lane admission requires transaction abort support", nil)
	}

	owners := make([]persistence.ParentLaneOwnership, 0, len(ordered))
	var heartbeat ParentLaneHeartbeatController
	for _, key := range ordered {
		owner, err := ps.parentLaneAdmission.Acquire(ctx, key, persistence.ParentLaneAcquireOptions{})
		if err != nil {
			ps.releaseParentLaneAdmissions(ctx, owners)
			return err
		}
		owners = append(owners, owner)
		if heartbeat == nil {
			heartbeat, err = ps.parentLaneAdmission.StartHeartbeat(ctx, []persistence.ParentLaneOwnership{owner}, persistence.ParentLaneHeartbeatOptions{})
		} else {
			err = heartbeat.AddOwner(owner)
		}
		if err != nil {
			ps.releaseParentLaneAdmissions(ctx, owners)
			return err
		}
	}
	return fn(ctx)
}

func (ps *PersistenceService) releaseParentLaneAdmissions(ctx context.Context, owners []persistence.ParentLaneOwnership) {
	releaseTimeout := parentLaneAdmissionCleanupReleaseTimeout
	if releaseTimeout <= 0 {
		releaseTimeout = persistence.DefaultParentLaneReleaseTimeout
	}
	for _, owner := range owners {
		releaseCtx, cancel := context.WithTimeout(context.Background(), releaseTimeout)
		ok, err := ps.parentLaneAdmission.Release(releaseCtx, owner)
		cancel()
		event := parentLaneAdmissionCleanupEvent{laneCount: 1, keyHash: shortParentLaneAdmissionHash(owner.KeyHash), err: err}
		switch {
		case err != nil:
			event.result = "failure"
		case !ok:
			event.result = "stale_owner"
		default:
			event.result = "success"
		}
		if observeParentLaneAdmissionCleanup != nil {
			observeParentLaneAdmissionCleanup(ctx, event)
		}
	}
}

func logParentLaneAdmissionCleanup(ctx context.Context, event parentLaneAdmissionCleanupEvent) {
	if event.result == "success" && !perftiming.Enabled(ctx) {
		return
	}
	logEvent := perftiming.BaseEvent(ctx, "parent_lane_admission_cleanup", event.result, event.err).
		Str("action", "release").
		Int("lane_count", event.laneCount)
	if event.keyHash != "" {
		logEvent = logEvent.Str("key_hash", event.keyHash)
	}
	logEvent.Msg("Parent lane admission cleanup")
}

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