package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sync"
	"time"

	database "github.com/nexuscrm/db"
	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"
	zlog "github.com/rs/zerolog/log"
)

const (
	parentLaneTableName = "_System_ParentLane"

	DefaultParentLaneLeaseDuration            = 30 * time.Second
	DefaultParentLaneReleaseTimeout           = 2 * time.Second
	DefaultParentLaneCompletionReleaseTimeout = 5 * time.Second
	DefaultParentLaneReleaseRetryInterval     = 10 * time.Millisecond
	DefaultParentLaneAcquireTimeout           = DefaultParentLaneLeaseDuration + DefaultParentLaneReleaseTimeout
	DefaultParentLanePollInterval             = 100 * time.Millisecond
	parentLaneAcquireSlowThreshold            = 250 * time.Millisecond
)

var ErrParentLaneContention = errors.New("parent lane is held by another transaction")

type ParentLaneAcquireOptions struct {
	OwnerToken   string
	Lease        time.Duration
	WaitTimeout  time.Duration
	PollInterval time.Duration
}

type ParentLaneOwnership struct {
	Key        recordlocks.ParentKey
	KeyHash    string
	OwnerToken string
	// LeaseExpiresAt is advisory for callers/tests. The SQL backend uses DB-server
	// CURRENT_TIMESTAMP(6) for durable expiry predicates and lease extension.
	LeaseExpiresAt time.Time
}

type parentLaneAcquireEvent struct {
	result     string
	objectName string
	keyHash    string
	started    time.Time
	polls      int
	err        error
}

type parentLaneAcquireObserver func(context.Context, parentLaneAcquireEvent)

type ParentLaneStore struct {
	backend        parentLaneBackend
	clock          parentLaneClock
	observeAcquire parentLaneAcquireObserver

	acquireGatesMu sync.Mutex
	acquireGates   map[string]*parentLaneAcquireGate

	lease                 time.Duration
	waitTimeout           time.Duration
	pollInterval          time.Duration
	releaseTimeout        time.Duration
	releaseAttemptTimeout time.Duration
	releaseRetryInterval  time.Duration
}

type parentLaneAcquireGate struct {
	token chan struct{}
	refs  int
}

type parentLaneClock interface {
	Now() time.Time
	Sleep(context.Context, time.Duration) error
}

type realParentLaneClock struct{}

func (realParentLaneClock) Now() time.Time { return time.Now().UTC() }

func (realParentLaneClock) Sleep(ctx context.Context, d time.Duration) error {
	if d <= 0 {
		return nil
	}
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-timer.C:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

type parentLaneRow struct {
	Key            recordlocks.ParentKey
	KeyHash        string
	OwnerToken     string
	LeaseExpiresAt time.Time
	HeartbeatAt    time.Time
	CreatedAt      time.Time
}

type parentLaneBackend interface {
	tryInsert(context.Context, parentLaneRow, time.Duration) (bool, error)
	tryTakeover(context.Context, parentLaneRow, time.Time, time.Duration) (bool, error)
	heartbeat(context.Context, string, string, time.Time, time.Time, time.Duration) (bool, error)
	release(context.Context, string, string) (bool, error)
}

type parentLaneExecutor interface {
	ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
}

// parentLaneSQLBackend deliberately computes lease timestamps and stale-owner
// predicates with DB-server CURRENT_TIMESTAMP(6), not the process clock. The
// process clock is used only for bounded client-side wait/sleep loops.
type parentLaneSQLBackend struct {
	exec parentLaneExecutor
}

func NewParentLaneStore(db *database.TiDBConnection) *ParentLaneStore {
	var exec parentLaneExecutor
	if db != nil {
		exec = db
	}
	return newParentLaneStoreForBackend(parentLaneSQLBackend{exec: exec}, realParentLaneClock{})
}

func newParentLaneStoreForBackend(backend parentLaneBackend, clock parentLaneClock) *ParentLaneStore {
	if clock == nil {
		clock = realParentLaneClock{}
	}
	return &ParentLaneStore{
		backend:               backend,
		clock:                 clock,
		observeAcquire:        logParentLaneAcquire,
		lease:                 DefaultParentLaneLeaseDuration,
		waitTimeout:           DefaultParentLaneAcquireTimeout,
		pollInterval:          DefaultParentLanePollInterval,
		releaseTimeout:        DefaultParentLaneCompletionReleaseTimeout,
		releaseAttemptTimeout: DefaultParentLaneReleaseTimeout,
		releaseRetryInterval:  DefaultParentLaneReleaseRetryInterval,
	}
}

func (s *ParentLaneStore) Acquire(ctx context.Context, key recordlocks.ParentKey, opts ParentLaneAcquireOptions) (ParentLaneOwnership, error) {
	if s == nil || s.backend == nil {
		return ParentLaneOwnership{}, apperrors.NewInternalError("parent lane store is not configured", nil)
	}
	canonical, ok := recordlocks.CanonicalParentLaneKey(key)
	if !ok {
		return ParentLaneOwnership{}, apperrors.NewValidationError("ParentLane", "parent lane key requires object name and record ID")
	}
	keyHash, ok := recordlocks.ParentLaneKeyHash(canonical)
	if !ok {
		return ParentLaneOwnership{}, apperrors.NewValidationError("ParentLane", "parent lane key requires object name and record ID")
	}
	acquireEvent := parentLaneAcquireEvent{
		result:     "started",
		objectName: canonical.ObjectName,
		keyHash:    shortParentLaneHash(keyHash),
		started:    time.Now(),
	}
	logParentLaneAcquireStart(ctx, acquireEvent)
	defer func() {
		if s.observeAcquire != nil {
			s.observeAcquire(ctx, acquireEvent)
		}
	}()
	token := opts.OwnerToken
	if token == "" {
		generated, err := recordlocks.NewParentLaneOwnerToken()
		if err != nil {
			acquireEvent.result = "owner_token_failure"
			acquireEvent.err = err
			return ParentLaneOwnership{}, apperrors.NewInternalError("failed to generate parent lane owner token", err)
		}
		token = generated
	}
	lease := opts.Lease
	if lease <= 0 {
		lease = s.lease
	}
	waitTimeout := opts.WaitTimeout
	if waitTimeout == 0 {
		waitTimeout = s.waitTimeout
	}
	pollInterval := opts.PollInterval
	if pollInterval <= 0 {
		pollInterval = s.pollInterval
	}

	releaseAcquireGate, err := s.waitForAcquireGate(ctx, keyHash)
	if err != nil {
		acquireEvent.result = "local_gate_failure"
		acquireEvent.err = err
		return ParentLaneOwnership{}, err
	}
	defer releaseAcquireGate()
	start := s.clock.Now()
	deadline := start.Add(waitTimeout)
	for {
		now := s.clock.Now()
		ownership := ParentLaneOwnership{
			Key:            canonical,
			KeyHash:        keyHash,
			OwnerToken:     token,
			LeaseExpiresAt: now.Add(lease),
		}
		row := parentLaneRow{
			Key:            canonical,
			KeyHash:        keyHash,
			OwnerToken:     token,
			LeaseExpiresAt: ownership.LeaseExpiresAt,
			HeartbeatAt:    now,
			CreatedAt:      now,
		}
		inserted, err := s.backend.tryInsert(ctx, row, lease)
		if err != nil {
			acquireEvent.result = "insert_failure"
			acquireEvent.err = err
			return ParentLaneOwnership{}, apperrors.NewInternalError("failed to insert parent lane ownership", err)
		}
		if inserted {
			acquireEvent.result = "inserted"
			return ownership, nil
		}
		taken, err := s.backend.tryTakeover(ctx, row, now, lease)
		if err != nil {
			acquireEvent.result = "takeover_failure"
			acquireEvent.err = err
			return ParentLaneOwnership{}, apperrors.NewInternalError("failed to take over stale parent lane ownership", err)
		}
		if taken {
			acquireEvent.result = "taken_over"
			return ownership, nil
		}
		acquireEvent.polls++
		if waitTimeout <= 0 || !now.Before(deadline) {
			acquireEvent.result = "contention_timeout"
			acquireEvent.err = ErrParentLaneContention
			return ParentLaneOwnership{}, apperrors.NewLockContentionError(ErrParentLaneContention)
		}
		sleepFor := pollInterval
		if remaining := deadline.Sub(now); remaining < sleepFor {
			sleepFor = remaining
		}
		if err := s.clock.Sleep(ctx, sleepFor); err != nil {
			acquireEvent.result = "wait_cancelled"
			acquireEvent.err = err
			return ParentLaneOwnership{}, err
		}
	}
}

func logParentLaneAcquireStart(ctx context.Context, event parentLaneAcquireEvent) {
	zlog.Debug().
		Str("event", "parent_lane_acquire").
		Str("request_id", perftiming.RequestID(ctx)).
		Str("status", "started").
		Str("object", event.objectName).
		Str("key_hash", event.keyHash).
		Msg("Parent lane acquire started")
}

func logParentLaneAcquire(ctx context.Context, event parentLaneAcquireEvent) {
	elapsed := time.Since(event.started)
	if event.err == nil && event.polls == 0 && elapsed < parentLaneAcquireSlowThreshold && !perftiming.Enabled(ctx) {
		return
	}
	perftiming.EventWithStatus(ctx, "parent_lane_acquire", event.started, event.result, event.err).
		Str("object", event.objectName).
		Str("key_hash", event.keyHash).
		Int("polls", event.polls).
		Msg("Parent lane acquire completed")
}

func (s *ParentLaneStore) waitForAcquireGate(ctx context.Context, keyHash string) (func(), error) {
	s.acquireGatesMu.Lock()
	if s.acquireGates == nil {
		s.acquireGates = make(map[string]*parentLaneAcquireGate)
	}
	gate := s.acquireGates[keyHash]
	if gate == nil {
		gate = &parentLaneAcquireGate{token: make(chan struct{}, 1)}
		gate.token <- struct{}{}
		s.acquireGates[keyHash] = gate
	}
	gate.refs++
	s.acquireGatesMu.Unlock()

	dropRef := func() {
		s.acquireGatesMu.Lock()
		gate.refs--
		if gate.refs == 0 && s.acquireGates[keyHash] == gate {
			delete(s.acquireGates, keyHash)
		}
		s.acquireGatesMu.Unlock()
	}

	select {
	case <-gate.token:
	case <-ctx.Done():
		dropRef()
		return nil, ctx.Err()
	}

	return func() {
		gate.token <- struct{}{}
		dropRef()
	}, nil
}

func (s *ParentLaneStore) Heartbeat(ctx context.Context, owner ParentLaneOwnership, lease time.Duration) (ParentLaneOwnership, bool, error) {
	if s == nil || s.backend == nil {
		return ParentLaneOwnership{}, false, apperrors.NewInternalError("parent lane store is not configured", nil)
	}
	if lease <= 0 {
		lease = s.lease
	}
	now := s.clock.Now()
	next := owner
	next.LeaseExpiresAt = now.Add(lease)
	ok, err := s.backend.heartbeat(ctx, owner.KeyHash, owner.OwnerToken, now, next.LeaseExpiresAt, lease)
	if err != nil {
		return ParentLaneOwnership{}, false, apperrors.NewInternalError("failed to heartbeat parent lane ownership", err)
	}
	if !ok {
		return owner, false, nil
	}
	return next, true, nil
}

func (s *ParentLaneStore) Release(ctx context.Context, owner ParentLaneOwnership) (bool, error) {
	if s == nil || s.backend == nil {
		return false, apperrors.NewInternalError("parent lane store is not configured", nil)
	}
	ok, err := s.backend.release(ctx, owner.KeyHash, owner.OwnerToken)
	if err != nil {
		return false, apperrors.NewInternalError("failed to release parent lane ownership", err)
	}
	return ok, nil
}

func (s *ParentLaneStore) ReleaseOnTransactionCompletion(ctx context.Context, owner ParentLaneOwnership) bool {
	return ports.RegisterTransactionCompletionHook(ctx, func(ports.TransactionCompletionOutcome) {
		_, _ = s.releaseWithCompletionBudget(owner, s.releaseTimeout)
	})
}

func (s *ParentLaneStore) releaseWithCompletionBudget(owner ParentLaneOwnership, releaseTimeout time.Duration) (bool, error) {
	if releaseTimeout <= 0 {
		releaseTimeout = DefaultParentLaneCompletionReleaseTimeout
	}
	attemptTimeout := s.releaseAttemptTimeout
	if attemptTimeout <= 0 {
		attemptTimeout = DefaultParentLaneReleaseTimeout
	}
	retryInterval := s.releaseRetryInterval
	if retryInterval <= 0 {
		retryInterval = DefaultParentLaneReleaseRetryInterval
	}

	releaseCtx, cancel := context.WithTimeout(context.Background(), releaseTimeout)
	defer cancel()
	for {
		attemptCtx, attemptCancel := context.WithTimeout(releaseCtx, attemptTimeout)
		ok, err := s.Release(attemptCtx, owner)
		attemptCancel()
		if err == nil {
			return ok, nil
		}
		if !errors.Is(err, context.DeadlineExceeded) || releaseCtx.Err() != nil {
			return false, err
		}
		if err := (realParentLaneClock{}).Sleep(releaseCtx, retryInterval); err != nil {
			return false, err
		}
	}
}

func (b parentLaneSQLBackend) tryInsert(ctx context.Context, row parentLaneRow, lease time.Duration) (bool, error) {
	if b.exec == nil {
		return false, errors.New("parent lane SQL executor is nil")
	}
	result, err := b.exec.ExecContext(ctx, fmt.Sprintf(
		"INSERT IGNORE INTO `%s` (`KeyHash`, `ParentObjectName`, `ParentRecordId`, `OwnerToken`, `LeaseExpiresAt`, `LastHeartbeatAt`, `CreatedDate`, `LastModifiedDate`) VALUES (?, ?, ?, ?, TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)), CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))",
		parentLaneTableName,
	), row.KeyHash, row.Key.ObjectName, row.Key.RecordID, row.OwnerToken, leaseMicroseconds(lease))
	return sqlRowsChanged(result, err)
}

func (b parentLaneSQLBackend) tryTakeover(ctx context.Context, row parentLaneRow, _ time.Time, lease time.Duration) (bool, error) {
	if b.exec == nil {
		return false, errors.New("parent lane SQL executor is nil")
	}
	result, err := b.exec.ExecContext(ctx, fmt.Sprintf(
		"UPDATE `%s` SET `OwnerToken` = ?, `LeaseExpiresAt` = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)), `LastHeartbeatAt` = CURRENT_TIMESTAMP(6), `LastModifiedDate` = CURRENT_TIMESTAMP(6) WHERE `KeyHash` = ? AND `ParentObjectName` = ? AND `ParentRecordId` = ? AND `LeaseExpiresAt` <= CURRENT_TIMESTAMP(6)",
		parentLaneTableName,
	), row.OwnerToken, leaseMicroseconds(lease), row.KeyHash, row.Key.ObjectName, row.Key.RecordID)
	return sqlRowsChanged(result, err)
}

func (b parentLaneSQLBackend) heartbeat(ctx context.Context, keyHash, ownerToken string, _, _ time.Time, lease time.Duration) (bool, error) {
	if b.exec == nil {
		return false, errors.New("parent lane SQL executor is nil")
	}
	result, err := b.exec.ExecContext(ctx, fmt.Sprintf(
		"UPDATE `%s` SET `LeaseExpiresAt` = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)), `LastHeartbeatAt` = CURRENT_TIMESTAMP(6), `LastModifiedDate` = CURRENT_TIMESTAMP(6) WHERE `KeyHash` = ? AND `OwnerToken` = ? AND `LeaseExpiresAt` > CURRENT_TIMESTAMP(6)",
		parentLaneTableName,
	), leaseMicroseconds(lease), keyHash, ownerToken)
	return sqlRowsChanged(result, err)
}

func (b parentLaneSQLBackend) release(ctx context.Context, keyHash, ownerToken string) (bool, error) {
	if b.exec == nil {
		return false, errors.New("parent lane SQL executor is nil")
	}
	result, err := b.exec.ExecContext(ctx, fmt.Sprintf(
		"DELETE FROM `%s` WHERE `KeyHash` = ? AND `OwnerToken` = ?",
		parentLaneTableName,
	), keyHash, ownerToken)
	return sqlRowsChanged(result, err)
}

func leaseMicroseconds(lease time.Duration) int64 {
	micros := lease.Microseconds()
	if micros <= 0 {
		return 1
	}
	return micros
}

func sqlRowsChanged(result sql.Result, err error) (bool, error) {
	if err != nil {
		return false, err
	}
	if result == nil {
		return false, nil
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return false, err
	}
	return rows > 0, nil
}