package mtprotoedge

import (
	"container/list"
	"context"
	"encoding/binary"
	"hash/maphash"
	"sync"
	"sync/atomic"
	"time"
)

const (
	// A valid client msg_id can be up to five minutes old or thirty seconds in
	// the future. The extra second covers scheduler and boundary jitter. This is
	// a no-ACK execution-receipt horizon, not a payload retention policy.
	rpcExecutionReceiptTTL = 331 * time.Second

	rpcExecutionMaxEntries        = 1 << 18
	rpcExecutionAuthMaxEntries    = 1 << 15
	rpcExecutionSessionMaxEntries = 1 << 14
	rpcExecutionPendingPerAuth    = 1 << 11

	// Receipts contain only fixed-shape identity/outcome metadata. This
	// conservative charge covers the receipt, list node, map bucket share and
	// reservation bookkeeping. Payload bytes are accounted exclusively by the
	// logical-session outbox.
	rpcExecutionReceiptBudgetBytes = 384

	// The complete replay identity is hashed with an instance-random seed. The
	// shard count is a power of two.
	rpcExecutionLedgerShards = 16
)

type rpcExecutionKey struct {
	authKeyID [8]byte
	sessionID int64
	reqMsgID  int64
}

// rpcReplayStore is the sole source of retained rpc_result payloads. The
// production implementation is SessionManager's logical-session outbox.
type rpcReplayStore interface {
	rpcResult(authKeyID [8]byte, sessionID, reqMsgID int64) (*encodedOutboundMessage, bool)
}

type rpcExecutionReceipt struct {
	key            rpcExecutionKey
	expiresAt      time.Time
	identity       rpcResultRequestIdentity
	admissionSeq   uint64
	executionKnown bool
	executionOK    bool
	// acknowledged exists only for the ACK-before-Complete race. A completed
	// receipt is removed immediately when the ACK wins.
	acknowledged bool
	// unavailable is a bounded execution tombstone. It prevents a completed
	// business operation from running again when no exact outbox frame exists.
	unavailable bool
	// The pending owner transfers this same global/auth/session entry
	// reservation to the completed receipt.
	reservation *rpcExecutionBudgetReservation
}

type rpcResultDependency struct {
	waiter    *rpcResultWaiter
	completed bool
	success   bool
}

// rpcExecutionLedger owns request execution identity and duplicate
// coordination. It never owns or copies rpc_result payload bytes; replay bytes
// are resolved from replayStore only while the logical-session outbox retains
// the exact unacknowledged frame.
type rpcExecutionLedger struct {
	shards              [rpcExecutionLedgerShards]rpcExecutionLedgerShard
	hashSeed            maphash.Seed
	reservedEntries     rpcResultFlightLimit
	receiptCount        atomic.Int64
	fairBudget          *rpcExecutionFairBudget
	flightLimit         rpcResultFlightLimit
	subscriberBudget    *rpcResultSubscriberBudget
	subscriberPerFlight int
	replayStore         rpcReplayStore
	nextAdmissionSeq    atomic.Uint64
	activeAdmissions    rpcAdmissionTracker
}

func (l *rpcExecutionLedger) stableAdmissionSafeFloor() uint64 {
	if l == nil {
		return 0
	}
	return l.activeAdmissions.stableSafeFloor(&l.nextAdmissionSeq)
}

type rpcExecutionLedgerShard struct {
	mu           sync.Mutex
	now          func() time.Time
	ttl          time.Duration
	receiptCount *atomic.Int64
	order        *list.List
	byKey        map[rpcExecutionKey]*list.Element
	// In-flight owners are independent from receipt TTL and cannot disappear
	// under completed-receipt pressure.
	pending map[rpcExecutionKey]*rpcResultFlight
}

type rpcExecutionLedgerCapacity struct {
	maxPending             int
	maxPendingPerAuth      int
	globalMaxEntries       int
	authMaxEntries         int
	sessionMaxEntries      int
	subscriberMaxGlobal    int
	subscriberMaxAuth      int
	subscriberMaxSession   int
	subscriberMaxPerFlight int
	replayStore            rpcReplayStore
}

func newRPCExecutionLedger(now func() time.Time, capacity rpcExecutionLedgerCapacity) *rpcExecutionLedger {
	if now == nil {
		now = time.Now
	}
	if capacity.replayStore == nil {
		panic("mtprotoedge: rpc execution ledger requires a replay store")
	}
	if capacity.maxPending <= 0 {
		capacity.maxPending = rpcResultFlightDefaultMaxPending
	}
	if capacity.maxPendingPerAuth <= 0 {
		capacity.maxPendingPerAuth = capacity.maxPending
	}
	if capacity.globalMaxEntries <= 0 {
		capacity.globalMaxEntries = rpcExecutionMaxEntries
	}
	if capacity.authMaxEntries <= 0 {
		capacity.authMaxEntries = capacity.globalMaxEntries
	}
	if capacity.sessionMaxEntries <= 0 {
		capacity.sessionMaxEntries = capacity.authMaxEntries
	}
	if capacity.subscriberMaxGlobal <= 0 {
		capacity.subscriberMaxGlobal = rpcResultSubscriberMaxGlobal
	}
	if capacity.subscriberMaxAuth <= 0 {
		capacity.subscriberMaxAuth = rpcResultSubscriberMaxAuth
	}
	if capacity.subscriberMaxSession <= 0 {
		capacity.subscriberMaxSession = rpcResultSubscriberMaxSession
	}
	if capacity.subscriberMaxPerFlight <= 0 {
		capacity.subscriberMaxPerFlight = rpcResultSubscriberMaxPerFlight
	}

	l := &rpcExecutionLedger{hashSeed: maphash.MakeSeed(), replayStore: capacity.replayStore}
	l.reservedEntries.max = int64(capacity.globalMaxEntries)
	l.flightLimit.max = int64(capacity.maxPending)
	l.fairBudget = newRPCExecutionFairBudget(
		l.hashSeed,
		&l.reservedEntries,
		int64(capacity.authMaxEntries),
		int64(capacity.sessionMaxEntries),
		capacity.maxPendingPerAuth,
	)
	l.subscriberBudget = newRPCResultSubscriberBudget(
		l.hashSeed,
		capacity.subscriberMaxGlobal,
		capacity.subscriberMaxAuth,
		capacity.subscriberMaxSession,
	)
	l.subscriberPerFlight = capacity.subscriberMaxPerFlight
	for i := range l.shards {
		s := &l.shards[i]
		s.now = now
		s.ttl = rpcExecutionReceiptTTL
		s.receiptCount = &l.receiptCount
		s.order = list.New()
		s.byKey = make(map[rpcExecutionKey]*list.Element)
		s.pending = make(map[rpcExecutionKey]*rpcResultFlight)
	}
	return l
}

func (l *rpcExecutionLedger) shard(key rpcExecutionKey) *rpcExecutionLedgerShard {
	return &l.shards[l.shardIndex(key)]
}

func (l *rpcExecutionLedger) shardIndex(key rpcExecutionKey) uint64 {
	var raw [24]byte
	copy(raw[:8], key.authKeyID[:])
	binary.LittleEndian.PutUint64(raw[8:16], uint64(key.sessionID))
	binary.LittleEndian.PutUint64(raw[16:24], uint64(key.reqMsgID))
	return maphash.Bytes(l.hashSeed, raw[:]) & (rpcExecutionLedgerShards - 1)
}

// Replay resolves an immutable result descriptor from the logical-session
// outbox. A receipt hit without an outbox frame is never treated as permission
// to execute the business handler again.
func (l *rpcExecutionLedger) Replay(authKeyID [8]byte, sessionID, reqMsgID int64) (*encodedOutboundMessage, bool) {
	if l == nil || reqMsgID == 0 {
		return nil, false
	}
	key := rpcExecutionKey{authKeyID: authKeyID, sessionID: sessionID, reqMsgID: reqMsgID}
	s := l.shard(key)
	now := s.now()
	s.mu.Lock()
	elem := s.byKey[key]
	if elem == nil {
		s.mu.Unlock()
		return nil, false
	}
	receipt := elem.Value.(*rpcExecutionReceipt)
	if !receipt.expiresAt.After(now) {
		s.removeElement(elem)
		s.mu.Unlock()
		return nil, false
	}
	if receipt.unavailable || receipt.acknowledged {
		s.mu.Unlock()
		return nil, false
	}
	s.mu.Unlock()
	return l.replayStore.rpcResult(authKeyID, sessionID, reqMsgID)
}

// Acknowledge removes a completed receipt immediately. reqMsgID must already
// have been resolved from the outbound actor's trusted server-msg-id mapping.
// A pending flight records the ACK so a racing completion cannot resurrect a
// receipt after the outbox body has been released.
func (l *rpcExecutionLedger) Acknowledge(authKeyID [8]byte, sessionID, reqMsgID int64) bool {
	if l == nil || reqMsgID == 0 {
		return false
	}
	key := rpcExecutionKey{authKeyID: authKeyID, sessionID: sessionID, reqMsgID: reqMsgID}
	s := l.shard(key)
	s.mu.Lock()
	s.expireLocked(s.now())
	if elem := s.byKey[key]; elem != nil {
		s.removeElement(elem)
		s.mu.Unlock()
		return true
	}
	if flight := s.pending[key]; flight != nil {
		flight.acknowledged = true
		s.mu.Unlock()
		return true
	}
	s.mu.Unlock()
	return false
}

// ObserveDependency returns a waiter for an admitted in-flight dependency, a
// completed execution outcome, or ok=false for an unknown request. It never
// creates execution ownership.
func (l *rpcExecutionLedger) ObserveDependency(authKeyID [8]byte, sessionID, reqMsgID int64) (rpcResultDependency, bool) {
	if l == nil || reqMsgID == 0 {
		return rpcResultDependency{}, false
	}
	key := rpcExecutionKey{authKeyID: authKeyID, sessionID: sessionID, reqMsgID: reqMsgID}
	s := l.shard(key)
	now := s.now()
	s.mu.Lock()
	defer s.mu.Unlock()
	if elem := s.byKey[key]; elem != nil {
		receipt := elem.Value.(*rpcExecutionReceipt)
		if receipt.expiresAt.After(now) {
			if !receipt.executionKnown {
				return rpcResultDependency{}, false
			}
			return rpcResultDependency{completed: true, success: receipt.executionOK}, true
		}
		s.removeElement(elem)
	}
	if flight := s.pending[key]; flight != nil {
		if flight.executionDone {
			return rpcResultDependency{completed: true, success: flight.executionOK}, true
		}
		return rpcResultDependency{waiter: &rpcResultWaiter{ledger: l, key: key, flight: flight}}, true
	}
	return rpcResultDependency{}, false
}

// Complete publishes terminal execution metadata and resolves current
// waiters. replayable is only a claim from the egress path; the ledger verifies
// that replayStore actually owns the exact frame before publishing a replayable
// receipt. encoded is passed transiently to joined waiters and is never stored.
func (l *rpcExecutionLedger) Complete(
	authKeyID [8]byte,
	sessionID, reqMsgID int64,
	encoded *encodedOutboundMessage,
	replayable bool,
) {
	if l == nil || reqMsgID == 0 || encoded == nil {
		return
	}
	if replayable {
		_, replayable = l.replayStore.rpcResult(authKeyID, sessionID, reqMsgID)
	}
	if l.completeOnce(authKeyID, sessionID, reqMsgID, encoded, replayable) {
		return
	}
	// An expired receipt in another shard may be the only global blocker.
	l.expireReceipts()
	_ = l.completeOnce(authKeyID, sessionID, reqMsgID, encoded, replayable)
}

// completeOnce returns false only when a cross-shard expiry reap may release
// the global reservation required by a defensive completion without an owner.
func (l *rpcExecutionLedger) completeOnce(
	authKeyID [8]byte,
	sessionID, reqMsgID int64,
	encoded *encodedOutboundMessage,
	replayable bool,
) bool {
	key := rpcExecutionKey{authKeyID: authKeyID, sessionID: sessionID, reqMsgID: reqMsgID}
	s := l.shard(key)
	s.mu.Lock()
	now := s.now()
	s.expireLocked(now)
	old := s.byKey[key]
	flight := s.pending[key]
	var oldReceipt *rpcExecutionReceipt
	if old != nil {
		oldReceipt = old.Value.(*rpcExecutionReceipt)
		if flight == nil {
			// Terminal publication is immutable. Late callbacks cannot extend TTL,
			// replace replay identity or resurrect an acknowledged result.
			s.mu.Unlock()
			return true
		}
	}

	identity, admissionSeq, executionKnown, executionOK, acknowledged := rpcResultFlightMetadataLocked(s, key)
	var reservation *rpcExecutionBudgetReservation
	switch {
	case flight != nil:
		reservation = flight.reservation
		if reservation == nil {
			s.mu.Unlock()
			panic("mtprotoedge: pending rpc execution has no fair-budget reservation")
		}
	case oldReceipt != nil && oldReceipt.reservation != nil:
		reservation = oldReceipt.reservation
	default:
		reservation = l.fairBudget.reserveCompleted(key)
		if reservation == nil {
			s.mu.Unlock()
			return false
		}
	}

	if old != nil {
		s.unlinkElement(old)
		if oldReceipt.reservation != nil && oldReceipt.reservation != reservation {
			oldReceipt.reservation.release()
			oldReceipt.reservation = nil
		}
	}
	receipt := &rpcExecutionReceipt{
		key:            key,
		expiresAt:      now.Add(s.ttl),
		identity:       identity,
		admissionSeq:   admissionSeq,
		executionKnown: executionKnown,
		executionOK:    executionOK,
		acknowledged:   acknowledged,
		unavailable:    !replayable,
		reservation:    reservation,
	}
	elem := s.order.PushBack(receipt)
	s.byKey[key] = elem
	s.incrementReceiptCount()

	subscribers, executionSubscribers, terminalExecutionOK := l.completeRPCResultFlightLocked(s, key, encoded)
	if acknowledged {
		// ACK won before completion. Current subscribers still receive encoded,
		// but no post-ACK receipt survives this critical section.
		s.removeElement(elem)
	}
	s.mu.Unlock()
	for _, subscriber := range subscribers {
		subscriber(encoded, true)
	}
	for _, subscriber := range executionSubscribers {
		subscriber(terminalExecutionOK)
	}
	return true
}

func rpcResultFlightMetadataLocked(s *rpcExecutionLedgerShard, key rpcExecutionKey) (
	rpcResultRequestIdentity,
	uint64,
	bool,
	bool,
	bool,
) {
	if flight := s.pending[key]; flight != nil {
		return flight.identity, flight.admissionSeq, flight.executionDone, flight.executionOK, flight.acknowledged
	}
	return rpcResultRequestIdentity{}, 0, false, false, false
}

func (l *rpcExecutionLedger) expireReceipts() {
	if l == nil {
		return
	}
	for i := range l.shards {
		s := &l.shards[i]
		s.mu.Lock()
		s.expireLocked(s.now())
		s.mu.Unlock()
	}
}

func (l *rpcExecutionLedger) receiptBudgetBytes() int64 {
	if l == nil {
		return 0
	}
	return l.receiptCount.Load() * rpcExecutionReceiptBudgetBytes
}

func (l *rpcExecutionLedger) Close() error { return nil }

func (l *rpcExecutionLedger) CloseContext(context.Context) error { return nil }

// forgetSession removes every terminal receipt for a destroyed logical
// session. SessionManager calls it only after physical producers converge.
func (l *rpcExecutionLedger) forgetSession(authKeyID [8]byte, sessionID int64) {
	if l == nil {
		return
	}
	for i := range l.shards {
		s := &l.shards[i]
		s.mu.Lock()
		for elem := s.order.Front(); elem != nil; {
			next := elem.Next()
			receipt := elem.Value.(*rpcExecutionReceipt)
			if receipt.key.authKeyID == authKeyID && receipt.key.sessionID == sessionID {
				s.removeElement(elem)
			}
			elem = next
		}
		s.mu.Unlock()
	}
}

func (s *rpcExecutionLedgerShard) expireLocked(now time.Time) {
	for elem := s.order.Front(); elem != nil; {
		next := elem.Next()
		receipt := elem.Value.(*rpcExecutionReceipt)
		if receipt.expiresAt.After(now) {
			return
		}
		s.removeElement(elem)
		elem = next
	}
}

func (s *rpcExecutionLedgerShard) removeElement(elem *list.Element) {
	receipt := s.unlinkElement(elem)
	if receipt != nil && receipt.reservation != nil {
		receipt.reservation.release()
		receipt.reservation = nil
	}
}

func (s *rpcExecutionLedgerShard) unlinkElement(elem *list.Element) *rpcExecutionReceipt {
	if elem == nil {
		return nil
	}
	receipt := elem.Value.(*rpcExecutionReceipt)
	delete(s.byKey, receipt.key)
	s.order.Remove(elem)
	s.decrementReceiptCount()
	return receipt
}

func (s *rpcExecutionLedgerShard) incrementReceiptCount() {
	if s.receiptCount == nil {
		panic("mtprotoedge: rpc execution receipt counter is unavailable")
	}
	s.receiptCount.Add(1)
}

func (s *rpcExecutionLedgerShard) decrementReceiptCount() {
	if s.receiptCount == nil || s.receiptCount.Add(-1) < 0 {
		panic("mtprotoedge: rpc execution receipt counter underflow")
	}
}