package mtprotoedge

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"time"

	"go.uber.org/zap"

	"github.com/iamxvbaba/td/bin"
	"github.com/iamxvbaba/td/mt"
	"github.com/iamxvbaba/td/proto"
	"github.com/iamxvbaba/td/tlprofile"
)

type inboundItemKind uint8

const (
	inboundItemDuplicate inboundItemKind = iota + 1
	inboundItemServiceDuplicate
	inboundItemPing
	inboundItemPingDelay
	inboundItemFutureSalts
	inboundItemMsgsAck
	inboundItemStateReq
	inboundItemResendReq
	inboundItemStateInfo
	inboundItemAllInfo
	inboundItemDestroySession
	inboundItemHTTPWait
	inboundItemDropAnswer
	inboundItemDestroyAuthKey
	inboundItemRPC
	inboundItemCapacityError
	inboundItemRPCAdmissionError
	// inboundItemRewrappedRPC is an initConnection retry whose exact inner TL
	// request is already executing (or completed) under the client's old msg_id.
	// It never dispatches business code a second time.
	inboundItemRewrappedRPC
	// inboundItemReplayRPC is a request first observed by this physical Conn whose
	// terminal result already exists in the cross-connection execution ledger. It is distinct
	// from inboundItemDuplicate: a duplicate already present in this Conn's seen
	// table must only be ACKed. The original owner/result is already using the same
	// reliable TCP stream, so replaying it once per retransmit wave amplifies a
	// client retry burst and can starve newer request IDs behind orphan results.
	inboundItemReplayRPC
)

type inboundItem struct {
	kind                          inboundItemKind
	msgID                         int64
	admissionSeq                  uint64
	seqNo                         int32
	typeID                        uint32
	content                       bool
	body                          []byte
	payload                       any
	admitted                      tlprofile.Admission
	method                        string
	replayAfterSuccessfulDelivery func() error
	layerProfileEvidenceFreshness inboundLayerProfileEvidenceFreshness
}

type inboundLayerProfileEvidenceFreshness uint8

const (
	// Unspecified is retained for focused force-style unit tests which construct
	// inboundItem directly, outside MTProto envelope preflight. Production items
	// are always classified from the frame's one clock sample.
	inboundLayerProfileEvidenceFreshnessUnspecified inboundLayerProfileEvidenceFreshness = iota
	inboundLayerProfileEvidenceFresh
	inboundLayerProfileEvidenceRequestBound
)

func (i inboundItem) profileEvidenceFresh() bool {
	return i.layerProfileEvidenceFreshness != inboundLayerProfileEvidenceRequestBound
}

type stagedClientMessage struct {
	msgID   int64
	seqNo   int32
	content bool
	service bool
}

type inboundPlan struct {
	items      []inboundItem
	staged     []stagedClientMessage
	ackIDs     []int64
	logicalMin int64
	releases   []func()
	// gzipExpandedBytes is the non-refundable per-frame decompression work
	// already performed by outer and exact-layer nested gzip envelopes. Memory
	// reservations are released when their buffers die, but this cumulative
	// counter prevents sibling RPCs from recycling the same CPU budget.
	gzipExpandedBytes int

	rpcReservation *inboundRPCBatchReservation
	rpcTasks       []inboundRPC
	rpcOwners      []*rpcResultOwnerLease
	rewrapAliases  []*rpcRewrapAlias
}

func (p *inboundPlan) close() {
	if p == nil {
		return
	}
	// Drop exact typed request graphs and uncommitted task closures before their
	// materialization reservation becomes reusable. Otherwise an abort could
	// advertise the same bytes to another connection while this plan still kept
	// the old graph reachable until its caller returned.
	for i := range p.items {
		p.items[i].admitted = tlprofile.Admission{}
	}
	for i := range p.rpcTasks {
		p.rpcTasks[i] = inboundRPC{}
	}
	p.rpcTasks = nil
	if p.rpcReservation != nil {
		p.rpcReservation.abort()
		p.rpcReservation = nil
	}
	for _, owner := range p.rpcOwners {
		owner.Abort()
	}
	p.rpcOwners = nil
	for _, alias := range p.rewrapAliases {
		alias.releaseCandidate()
		if alias != nil && alias.newOwner != nil {
			alias.newOwner.Abort()
		}
	}
	p.rewrapAliases = nil
	for i := len(p.releases) - 1; i >= 0; i-- {
		p.releases[i]()
	}
	p.releases = nil
}

func (p *inboundPlan) commitRewrapAliases(s *Server) error {
	if p == nil || len(p.rewrapAliases) == 0 {
		return nil
	}
	for i, alias := range p.rewrapAliases {
		if err := alias.activate(s); err != nil {
			for _, pending := range p.rewrapAliases[i:] {
				pending.releaseCandidate()
				if pending != nil && pending.newOwner != nil {
					pending.newOwner.Abort()
				}
			}
			p.rewrapAliases = nil
			return err
		}
	}
	p.rewrapAliases = nil
	return nil
}

// rejectNewRPCOwners turns only ownership acquired by this batch into bounded
// capacity responses. Existing completed replays and pending joins remain
// active: canceling them would either lose a response or publish into another
// request's flight.
func (p *inboundPlan) rejectNewRPCOwners(indices []int) {
	if p == nil {
		return
	}
	for _, index := range indices {
		if index >= 0 && index < len(p.items) {
			p.items[index].kind = inboundItemCapacityError
		}
	}
	kept := p.rewrapAliases[:0]
	for _, alias := range p.rewrapAliases {
		if alias == nil || alias.newOwner == nil {
			kept = append(kept, alias)
			continue
		}
		if alias.itemIndex >= 0 && alias.itemIndex < len(p.items) {
			p.items[alias.itemIndex].kind = inboundItemCapacityError
			p.items[alias.itemIndex].payload = nil
		}
		alias.releaseCandidate()
		// This owner was acquired by the rejected batch and is not present in
		// plan.rpcOwners because it belonged to a rewrap alias. Abort it here
		// before dropping the alias, otherwise the exact flight remains pending
		// forever with no task or publisher able to complete it.
		alias.newOwner.Abort()
		alias.newOwner = nil
	}
	p.rewrapAliases = kept
}

func (p *inboundPlan) commitRPCBatch() error {
	if p == nil || p.rpcReservation == nil {
		return nil
	}
	// handleEncrypted calls this only after ownership/session-control barriers,
	// connState commit and every synchronous service action have completed. The
	// batch is runnable immediately; using the old deferred scheduler token here
	// was not a real barrier on a busy Conn because an existing ready token could
	// dequeue newly appended tasks before activateRPCBatch ran.
	err := p.rpcReservation.commit(p.rpcTasks)
	if err != nil {
		return err
	}
	p.rpcReservation = nil
	p.rpcTasks = nil
	// Every owner is now attached to exactly one queued task. Its task.release
	// path aborts if no terminal rpc_result was published.
	p.rpcOwners = nil
	return nil
}

func (p *inboundPlan) includeLogicalID(msgID int64) {
	if p.logicalMin == 0 || msgID < p.logicalMin {
		p.logicalMin = msgID
	}
}

func (p *inboundPlan) commitState(cs *connState) {
	for _, m := range p.staged {
		cs.trackInbound(m.msgID, m.seqNo, m.content, m.service, msgStateReceived)
	}
}

type connStateOverlay struct {
	base            *connState
	staged          []stagedClientMessage
	maxContentMsgID int64
	maxContentSeqNo int32
}

func newConnStateOverlay(base *connState) connStateOverlay {
	return connStateOverlay{
		base:            base,
		maxContentMsgID: base.maxContentMsgID,
		maxContentSeqNo: base.maxContentSeqNo,
	}
}

func (o *connStateOverlay) seenRecord(msgID int64) (clientMsgRecord, bool) {
	for i := len(o.staged) - 1; i >= 0; i-- {
		m := o.staged[i]
		if m.msgID == msgID {
			return clientMsgRecord{state: msgStateReceived, seqNo: m.seqNo, content: m.content, service: m.service}, true
		}
	}
	return o.base.seenRecord(msgID)
}

func (o *connStateOverlay) validateSeq(msgID int64, seqNo int32, content bool) int {
	if !content {
		return 0
	}
	if msgID > o.maxContentMsgID && seqNo > o.maxContentSeqNo {
		return 0
	}
	for seenMsgID, record := range o.base.seen {
		if !record.content {
			continue
		}
		if seenMsgID < msgID && record.seqNo >= seqNo {
			return badMsgSeqTooLow
		}
		if seenMsgID > msgID && record.seqNo <= seqNo {
			return badMsgSeqTooHigh
		}
	}
	for _, record := range o.staged {
		if !record.content {
			continue
		}
		if record.msgID < msgID && record.seqNo >= seqNo {
			return badMsgSeqTooLow
		}
		if record.msgID > msgID && record.seqNo <= seqNo {
			return badMsgSeqTooHigh
		}
	}
	return 0
}

func (o *connStateOverlay) stage(msgID int64, seqNo int32, content, service bool) {
	o.staged = append(o.staged, stagedClientMessage{msgID: msgID, seqNo: seqNo, content: content, service: service})
	if content {
		if msgID > o.maxContentMsgID {
			o.maxContentMsgID = msgID
		}
		if seqNo > o.maxContentSeqNo {
			o.maxContentSeqNo = seqNo
		}
	}
}

type inboundScope struct {
	insideContainer bool
	mustBeDuplicate bool
}

type inboundPreflightBudget struct {
	depth          int
	containerDepth int
	expanded       int
	now            time.Time
}

func (s *Server) preflightInbound(cs *connState, msgID int64, seqNo int32, body []byte) (*inboundPlan, error) {
	plan := &inboundPlan{logicalMin: 0}
	overlay := newConnStateOverlay(cs)
	// All envelope checks in one transport frame use the same clock sample. Apart
	// from making boundary behavior deterministic, this lets walkInbound reject
	// an invalid outer msg_id before spending CPU or memory on gzip expansion.
	budget := &inboundPreflightBudget{now: s.clock.Now()}
	if err := s.walkInbound(plan, &overlay, msgID, seqNo, body, inboundScope{}, budget); err != nil {
		plan.close()
		return nil, err
	}
	plan.gzipExpandedBytes = budget.expanded
	plan.staged = overlay.staged
	if plan.logicalMin == 0 {
		plan.close()
		return nil, fmt.Errorf("inbound plan accepted no logical message")
	}
	for _, item := range plan.items {
		if item.kind == inboundItemDestroyAuthKey && len(plan.items) != 1 {
			plan.close()
			return nil, &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
		}
	}
	return plan, nil
}

func (s *Server) walkInbound(
	plan *inboundPlan,
	overlay *connStateOverlay,
	msgID int64,
	seqNo int32,
	body []byte,
	scope inboundScope,
	budget *inboundPreflightBudget,
) error {
	if budget.depth > maxDispatchDepth {
		return fmt.Errorf("mtproto wrapper depth %d exceeds %d", budget.depth, maxDispatchDepth)
	}
	b := &bin.Buffer{Buf: body}
	typeID, err := b.PeekID()
	if err != nil {
		return fmt.Errorf("peek type id: %w", err)
	}
	if code := validateInboundMessageID(budget.now, msgID, scope.insideContainer); code != 0 {
		if scope.insideContainer {
			code = badMsgContainer
		}
		return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
	}

	// A repeated outer container still has to be decoded far enough to enumerate
	// its inner request ids, but none of those already-accepted inner bodies needs
	// decoding again. In particular, never reinflate a gzip body merely to discover
	// the content bit we already retained with its msg_id. If an outer duplicate
	// introduces an unseen inner id, reject the container before touching its body.
	// A seen top-level content gzip is also safe to short-circuit: a container is
	// non-content, so it cannot be hidden behind that retained record. Ambiguous
	// non-content top-level gzip still expands to distinguish a container wrapper.
	if record, seen := overlay.seenRecord(msgID); scope.mustBeDuplicate ||
		(typeID == proto.GZIPTypeID && seen && (scope.insideContainer || record.content)) {
		if !seen || record.seqNo != seqNo {
			return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
		}
		return appendInboundDuplicate(plan, msgID, seqNo, typeID, record)
	}
	if typeID == proto.GZIPTypeID {
		data, release, err := s.decodeGZIPWithGlobalBudget(b)
		if err != nil {
			return fmt.Errorf("decode gzip: %w", err)
		}
		plan.releases = append(plan.releases, release)
		budget.expanded += len(data)
		if budget.expanded > maxDispatchExpandedBytes {
			return fmt.Errorf("cumulative gzip expansion %d exceeds %d", budget.expanded, maxDispatchExpandedBytes)
		}
		budget.depth++
		err = s.walkInbound(plan, overlay, msgID, seqNo, data, scope, budget)
		budget.depth--
		return err
	}

	if typeID == proto.MessageContainerTypeID {
		if scope.insideContainer || budget.containerDepth != 0 {
			return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
		}
		if code := validateClientEnvelope(budget.now, msgID, seqNo, typeID); code != 0 {
			return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
		}
		outerRecord, outerSeen := overlay.seenRecord(msgID)
		if outerSeen {
			if outerRecord.content || outerRecord.seqNo != seqNo {
				return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
			}
		} else {
			if code := overlay.validateSeq(msgID, seqNo, false); code != 0 {
				return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
			}
			overlay.stage(msgID, seqNo, false, false)
		}

		count, err := containerMessageCount(b)
		if err != nil {
			return fmt.Errorf("decode container count: %w", err)
		}
		if count > maxContainerMessages {
			return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
		}
		container, release, err := s.decodeMessageContainerViews(b, count)
		if err != nil {
			return fmt.Errorf("decode container: %w", err)
		}
		plan.releases = append(plan.releases, release)
		plan.items = growInboundItems(plan.items, count)
		plan.ackIDs = growInt64s(plan.ackIDs, count)
		overlay.staged = growStagedMessages(overlay.staged, count+1)
		if len(container.Messages) == 0 {
			plan.includeLogicalID(msgID)
			return nil
		}

		budget.depth++
		budget.containerDepth++
		for _, m := range container.Messages {
			if m.ID >= msgID || int32(m.SeqNo) > seqNo {
				return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
			}
			if err := s.walkInbound(plan, overlay, m.ID, int32(m.SeqNo), m.Body, inboundScope{
				insideContainer: true,
				mustBeDuplicate: outerSeen,
			}, budget); err != nil {
				return err
			}
		}
		budget.containerDepth--
		budget.depth--
		return nil
	}

	if scope.insideContainer {
		if code := validateClientContainerEnvelope(msgID, seqNo, typeID); code != 0 {
			return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
		}
	} else if code := validateClientEnvelope(budget.now, msgID, seqNo, typeID); code != 0 {
		return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
	}

	content := clientMessageIsContentRelated(typeID, seqNo)
	if record, seen := overlay.seenRecord(msgID); seen {
		if record.seqNo != seqNo || record.content != content {
			return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
		}
		return appendInboundDuplicate(plan, msgID, seqNo, typeID, record)
	}
	if scope.mustBeDuplicate {
		return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: badMsgContainer}
	}
	if code := overlay.validateSeq(msgID, seqNo, content); code != 0 {
		return &dispatchBadMsgError{msgID: msgID, seqNo: seqNo, code: code}
	}
	overlay.stage(msgID, seqNo, content, inboundTypeIsService(typeID))
	plan.includeLogicalID(msgID)

	item, err := preflightInboundItem(msgID, seqNo, typeID, content, body)
	if err != nil {
		return err
	}
	if validateInboundMessageID(budget.now, msgID, false) == 0 {
		item.layerProfileEvidenceFreshness = inboundLayerProfileEvidenceFresh
	} else {
		// Inner container messages deliberately bypass the wall-clock rejection
		// above, but old/future ids are request-bound and cannot publish mutable
		// Layer/init/readiness/auth-bind evidence.
		item.layerProfileEvidenceFreshness = inboundLayerProfileEvidenceRequestBound
	}
	plan.items = append(plan.items, item)
	if content {
		plan.ackIDs = append(plan.ackIDs, msgID)
	}
	return nil
}

func validateInboundMessageID(now time.Time, msgID int64, insideContainer bool) int {
	if !validClientMessageIDBits(msgID) {
		return badMsgIDInvalidBits
	}
	// A container's outer envelope supplies the wall-clock admission boundary for
	// its inner messages. Inner ids still need the client low bits checked before
	// wrapper expansion, but intentionally keep the established no-time-check rule.
	if insideContainer {
		return 0
	}
	msgTime := proto.MessageID(msgID).Time()
	if msgTime.Before(now.Add(-300 * time.Second)) {
		return badMsgIDTooLow
	}
	if msgTime.After(now.Add(30 * time.Second)) {
		return badMsgIDTooHigh
	}
	return 0
}

func validClientMessageIDBits(msgID int64) bool {
	return msgID > 0 && uint32(msgID) != 0 &&
		proto.MessageID(msgID).Type() == proto.MessageFromClient
}

func appendInboundDuplicate(plan *inboundPlan, msgID int64, seqNo int32, typeID uint32, record clientMsgRecord) error {
	plan.includeLogicalID(msgID)
	kind := inboundItemDuplicate
	if record.service {
		kind = inboundItemServiceDuplicate
	}
	plan.items = append(plan.items, inboundItem{
		kind: kind, msgID: msgID, seqNo: seqNo, typeID: typeID, content: record.content,
	})
	if record.content {
		plan.ackIDs = append(plan.ackIDs, msgID)
	}
	return nil
}

func inboundTypeIsService(typeID uint32) bool {
	switch typeID {
	case mt.PingRequestTypeID,
		mt.PingDelayDisconnectRequestTypeID,
		mt.GetFutureSaltsRequestTypeID,
		mt.MsgsAckTypeID,
		mt.MsgsStateReqTypeID,
		mt.MsgResendReqTypeID,
		mt.MsgsStateInfoTypeID,
		mt.MsgsAllInfoTypeID,
		mt.DestroySessionRequestTypeID,
		mt.HTTPWaitRequestTypeID,
		mt.RPCDropAnswerRequestTypeID,
		destroyAuthKeyRequestTypeID:
		return true
	default:
		return false
	}
}

func growInboundItems(items []inboundItem, extra int) []inboundItem {
	if extra <= cap(items)-len(items) {
		return items
	}
	grown := make([]inboundItem, len(items), len(items)+extra)
	copy(grown, items)
	return grown
}

func growStagedMessages(items []stagedClientMessage, extra int) []stagedClientMessage {
	if extra <= cap(items)-len(items) {
		return items
	}
	grown := make([]stagedClientMessage, len(items), len(items)+extra)
	copy(grown, items)
	return grown
}

func growInt64s(items []int64, extra int) []int64 {
	if extra <= cap(items)-len(items) {
		return items
	}
	grown := make([]int64, len(items), len(items)+extra)
	copy(grown, items)
	return grown
}

type stateInfoPayload struct {
	reqMsgID int64
	info     []byte
}

type allInfoPayload struct {
	count int
	info  []byte
}

// int64VectorView is a bounded, immutable view of a TL Vector<long>. Keeping the
// encoded bytes avoids one retained []int64 allocation per service message while
// a whole container is in preflight. Execution materializes the small bounded
// slice only for the existing Conn APIs that require []int64, after all wrappers
// and every sibling message have already passed structural validation.
type int64VectorView struct {
	raw   []byte
	count int
}

func decodeInt64VectorView(b *bin.Buffer, expectedTypeID uint32, max int) (int64VectorView, error) {
	if b == nil || len(b.Buf) < 12 {
		return int64VectorView{}, io.ErrUnexpectedEOF
	}
	if got := binary.LittleEndian.Uint32(b.Buf[:4]); got != expectedTypeID {
		return int64VectorView{}, fmt.Errorf("unexpected constructor %#x", got)
	}
	if got := binary.LittleEndian.Uint32(b.Buf[4:8]); got != bin.TypeVector {
		return int64VectorView{}, fmt.Errorf("unexpected vector constructor %#x", got)
	}
	count := int(int32(binary.LittleEndian.Uint32(b.Buf[8:12])))
	if count < 0 {
		return int64VectorView{}, fmt.Errorf("negative vector count %d", count)
	}
	if count > max {
		return int64VectorView{}, fmt.Errorf("vector count %d exceeds %d", count, max)
	}
	if count > (len(b.Buf)-12)/8 {
		return int64VectorView{}, io.ErrUnexpectedEOF
	}
	end := 12 + count*8
	if end != len(b.Buf) {
		return int64VectorView{}, fmt.Errorf("long vector has %d trailing bytes", len(b.Buf)-end)
	}
	return int64VectorView{raw: b.Buf[12:end:end], count: count}, nil
}

func (v int64VectorView) materialize() []int64 {
	if v.count == 0 {
		return nil
	}
	ids := make([]int64, v.count)
	for i := range ids {
		offset := i * 8
		ids[i] = int64(binary.LittleEndian.Uint64(v.raw[offset : offset+8]))
	}
	return ids
}

func decodeInboundServiceExact(b *bin.Buffer, decoder bin.Decoder) error {
	if err := decoder.Decode(b); err != nil {
		return err
	}
	if b.Len() != 0 {
		return fmt.Errorf("service message has %d trailing bytes", b.Len())
	}
	return nil
}

func preflightInboundItem(msgID int64, seqNo int32, typeID uint32, content bool, body []byte) (inboundItem, error) {
	item := inboundItem{msgID: msgID, seqNo: seqNo, typeID: typeID, content: content, body: body}
	b := &bin.Buffer{Buf: body}
	switch typeID {
	case mt.PingRequestTypeID:
		var value mt.PingRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, fmt.Errorf("decode ping: %w", err)
		}
		item.kind, item.payload = inboundItemPing, value
	case mt.PingDelayDisconnectRequestTypeID:
		var value mt.PingDelayDisconnectRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, fmt.Errorf("decode ping_delay_disconnect: %w", err)
		}
		item.kind, item.payload = inboundItemPingDelay, value
	case mt.GetFutureSaltsRequestTypeID:
		var value mt.GetFutureSaltsRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, fmt.Errorf("decode get_future_salts: %w", err)
		}
		item.kind, item.payload = inboundItemFutureSalts, value
	case mt.MsgsAckTypeID:
		value, err := decodeInt64VectorView(b, mt.MsgsAckTypeID, maxServiceMessageIDs)
		if err != nil {
			return item, fmt.Errorf("decode msgs_ack: %w", err)
		}
		item.kind, item.payload = inboundItemMsgsAck, value
	case mt.MsgsStateReqTypeID:
		value, err := decodeInt64VectorView(b, mt.MsgsStateReqTypeID, maxServiceMessageIDs)
		if err != nil {
			return item, fmt.Errorf("decode msgs_state_req: %w", err)
		}
		item.kind, item.payload = inboundItemStateReq, value
	case mt.MsgResendReqTypeID:
		value, err := decodeInt64VectorView(b, mt.MsgResendReqTypeID, maxServiceMessageIDs)
		if err != nil {
			return item, fmt.Errorf("decode msg_resend_req: %w", err)
		}
		item.kind, item.payload = inboundItemResendReq, value
	case mt.MsgsStateInfoTypeID:
		reqMsgID, info, err := msgsStateInfoView(b)
		if err != nil {
			return item, fmt.Errorf("decode msgs_state_info: %w", err)
		}
		item.kind, item.payload = inboundItemStateInfo, stateInfoPayload{reqMsgID: reqMsgID, info: info}
	case mt.MsgsAllInfoTypeID:
		count, info, err := msgsAllInfoView(b)
		if err != nil {
			return item, fmt.Errorf("decode msgs_all_info: %w", err)
		}
		if len(info) != count {
			return item, fmt.Errorf("decode msgs_all_info: info length %d does not match msg_ids %d", len(info), count)
		}
		item.kind, item.payload = inboundItemAllInfo, allInfoPayload{count: count, info: info}
	case mt.DestroySessionRequestTypeID:
		var value mt.DestroySessionRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, fmt.Errorf("decode destroy_session: %w", err)
		}
		item.kind, item.payload = inboundItemDestroySession, value
	case mt.HTTPWaitRequestTypeID:
		var value mt.HTTPWaitRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, fmt.Errorf("decode http_wait: %w", err)
		}
		item.kind, item.payload = inboundItemHTTPWait, value
	case mt.RPCDropAnswerRequestTypeID:
		var value mt.RPCDropAnswerRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, fmt.Errorf("decode rpc_drop_answer: %w", err)
		}
		item.kind, item.payload = inboundItemDropAnswer, value
	case destroyAuthKeyRequestTypeID:
		var value destroyAuthKeyRequest
		if err := decodeInboundServiceExact(b, &value); err != nil {
			return item, err
		}
		item.kind, item.payload = inboundItemDestroyAuthKey, value
	default:
		item.kind = inboundItemRPC
	}
	return item, nil
}

// prepareInboundRPCBatch performs the whole container's count/byte admission
// before copying or scheduling any API RPC. Capacity exhaustion is converted
// into one consistent local 500 WORKER_BUSY_TOO_LONG_RETRY result per new
// RPC; no business handler from the batch is allowed to start in that case.
func (s *Server) prepareInboundRPCBatch(ctx context.Context, c *Conn, plan *inboundPlan) error {
	if s.layerRPC != nil {
		return s.prepareInboundLayerRPCBatch(ctx, c, plan)
	}
	// Keep service-only frames (ping/ack/http_wait) allocation-free here. These
	// collections are needed only after the first real API RPC acquires ownership.
	var indices []int
	var specs []inboundRPCSpec
	var ownersInPlan map[int64]*rpcResultOwnerLease
	flightCapacity := false
	clearedPostInitCandidates := false
	for i := range plan.items {
		item := &plan.items[i]
		if item.kind != inboundItemRPC && item.kind != inboundItemDuplicate {
			continue
		}
		localDuplicate := item.kind == inboundItemDuplicate
		if localDuplicate {
			// connState already proves this msg_id was admitted on this physical
			// generation. Its original owner either still holds the flight or has
			// published a result after physical write; do not consume a global flight
			// slot merely to ACK the duplicate.
			continue
		}
		method := s.typeName(item.typeID)
		init, isInitRewrap := decodeRPCRewrapInit(item.body)
		if isInitRewrap {
			firstInit := false
			if item.profileEvidenceFresh() {
				firstInit = !c.rpcRewrapInitialized.Swap(true)
				c.setLegacyClientLayer(init.layer)
			}
			if candidate := s.rpcRewrap.claim(c, init.inner); candidate != nil {
				claim, err := s.rpcResults.Acquire(c.authKeyID, c.sessionID, item.msgID)
				if errors.Is(err, ErrRPCResultFlightCapacity) {
					s.rpcRewrap.release(candidate)
					c.metrics.InboundRPCDropped(candidate.method, "flight_capacity")
					flightCapacity = true
					item.kind = inboundItemCapacityError
					continue
				}
				if err != nil {
					s.rpcRewrap.release(candidate)
					return err
				}
				switch claim.state {
				case rpcResultAcquireCompleted:
					s.rpcRewrap.commit(candidate)
					item.kind = inboundItemReplayRPC
					item.payload = claim.encoded
				case rpcResultAcquireAcknowledged:
					// The client already ACKed the correlated rpc_result. Retire the
					// stale rewrap candidate and keep this duplicate ACK-only; neither
					// the old body nor the business handler may run again.
					s.rpcRewrap.commit(candidate)
					item.kind = inboundItemDuplicate
					item.payload = nil
				case rpcResultAcquirePending:
					s.rpcRewrap.commit(candidate)
					item.kind = inboundItemRewrappedRPC
					item.payload = claim.waiter
					plan.rewrapAliases = append(plan.rewrapAliases, &rpcRewrapAlias{
						conn: c, itemIndex: i, newReqID: item.msgID, method: candidate.method,
						oldWaiter: claim.waiter, observeInit: firstInit, init: init,
					})
				case rpcResultAcquireOwner:
					if ownersInPlan == nil {
						ownersInPlan = make(map[int64]*rpcResultOwnerLease)
					}
					ownersInPlan[item.msgID] = claim.owner
					item.kind = inboundItemRewrappedRPC
					item.payload = claim.owner
					plan.rewrapAliases = append(plan.rewrapAliases, &rpcRewrapAlias{
						conn: c, itemIndex: i, newReqID: item.msgID, method: candidate.method,
						oldWaiter: candidate.waiter, newOwner: claim.owner,
						sourceConn: candidate.source, sourceOwner: candidate.owner,
						observeInit: firstInit, init: init,
						candidate: candidate, registry: s.rpcRewrap,
					})
				default:
					s.rpcRewrap.release(candidate)
					return ErrRPCResultFlightInvalid
				}
				s.log.Info("RPC init rewrap matched",
					zap.String("method", candidate.method),
					zap.Int64("old_req_msg_id", candidate.reqMsgID),
					zap.Int64("new_req_msg_id", item.msgID),
					zap.Bool("same_connection", candidate.source == c),
					zap.String("auth_key_id", c.authKeyHex), zap.Int64("session_id", c.sessionID))
				continue
			}
		} else if item.profileEvidenceFresh() && c.rpcRewrapInitialized.Load() && !clearedPostInitCandidates {
			// A naked request after this connection has observed initConnection is
			// event-level proof that the client finished moving its old running set.
			// Retire any unmatched candidates without a timer.
			s.rpcRewrap.clearSession(c)
			clearedPostInitCandidates = true
		}
		claim, err := s.rpcResults.Acquire(c.authKeyID, c.sessionID, item.msgID)
		if errors.Is(err, ErrRPCResultFlightCapacity) {
			if item.kind == inboundItemRPC {
				c.metrics.InboundRPCDropped(method, "flight_capacity")
				flightCapacity = true
				item.kind = inboundItemCapacityError
			}
			continue
		}
		if err != nil {
			return err
		}
		switch claim.state {
		case rpcResultAcquireCompleted:
			s.log.Info("RPC duplicate replay from logical-session outbox",
				zap.String("method", method),
				zap.Int64("msg_id", item.msgID),
				zap.String("auth_key_id", c.authKeyHex),
				zap.Int64("session_id", c.sessionID),
			)
			item.kind = inboundItemReplayRPC
			item.payload = claim.encoded
		case rpcResultAcquireAcknowledged:
			item.kind = inboundItemDuplicate
			item.payload = nil
		case rpcResultAcquirePending:
			// A malformed/replayed container may repeat the same msg_id after this
			// very plan installed its owner. More generally, any request already in
			// this Conn's seen table shares the owner's reliable response path. Only
			// a fresh physical replacement may wait for and replay the old result.
			if ownersInPlan[item.msgID] != nil {
				item.kind = inboundItemDuplicate
				item.payload = nil
			} else {
				item.kind = inboundItemRewrappedRPC
				item.payload = claim.waiter
				plan.rewrapAliases = append(plan.rewrapAliases, &rpcRewrapAlias{
					conn: c, itemIndex: i, newReqID: item.msgID, method: method, oldWaiter: claim.waiter,
				})
			}
		case rpcResultAcquireOwner:
			if item.kind == inboundItemDuplicate {
				// connState says this request was already accepted, so absence from
				// both completed and in-flight tables is not authority to execute it.
				claim.owner.Abort()
				continue
			}
			if ownersInPlan == nil {
				ownersInPlan = make(map[int64]*rpcResultOwnerLease)
				indices = make([]int, 0, len(plan.items))
				specs = make([]inboundRPCSpec, 0, len(plan.items))
			}
			ownersInPlan[item.msgID] = claim.owner
			plan.rpcOwners = append(plan.rpcOwners, claim.owner)
			item.payload = claim.owner
			indices = append(indices, i)
			specs = append(specs, inboundRPCSpec{method: method, size: len(item.body)})
			if !c.rpcRewrapInitialized.Load() && !isInitRewrap {
				s.rpcRewrap.register(c, item.body, item.msgID, method, claim.owner)
			}
		default:
			return ErrRPCResultFlightInvalid
		}
	}
	if flightCapacity {
		// One container is one API admission unit. If the cross-connection
		// exactly-once table cannot claim every fresh request, none of this
		// batch may reach a business handler.
		for _, index := range indices {
			plan.items[index].kind = inboundItemCapacityError
		}
		plan.rejectNewRPCOwners(indices)
		return nil
	}
	if len(specs) == 0 {
		return nil
	}

	reservation, err := c.reserveInboundRPCBatch(ctx, specs)
	if err != nil {
		if errors.Is(err, ErrInboundRPCQueueFull) {
			for _, index := range indices {
				plan.items[index].kind = inboundItemCapacityError
			}
			plan.rejectNewRPCOwners(indices)
			return nil
		}
		return err
	}
	plan.rpcReservation = reservation
	plan.rpcTasks = make([]inboundRPC, len(indices))
	for i, index := range indices {
		item := &plan.items[index]
		body := append([]byte(nil), item.body...)
		owner, _ := item.payload.(*rpcResultOwnerLease)
		plan.rpcTasks[i] = s.newInboundRPCTask(c, item.msgID, specs[i].method, body, owner)
	}
	return nil
}

func (s *Server) executeInboundPlan(ctx context.Context, cs *connState, c *Conn, plan *inboundPlan) error {
	for _, item := range plan.items {
		switch item.kind {
		case inboundItemDuplicate:
			// The preflight plan already stages a content ACK for a locally seen
			// duplicate. Do not wait for or replay its result on the same reliable
			// stream; the original owner is solely responsible for that response.
			continue
		case inboundItemServiceDuplicate:
			// Classify from the originally committed connState record, never from the
			// retransmitted body. This prevents same-id payload replacement. Only an
			// already retained answer is eligible for resend (rpc_drop_answer today);
			// other best-effort service traffic uses a later fresh request.
			if err := s.replayRPCResultByRequest(ctx, c, item.msgID); err != nil {
				return err
			}
		case inboundItemReplayRPC:
			if encoded, _ := item.payload.(*encodedOutboundMessage); encoded != nil {
				if err := s.sendReplayedRPCResultWithHook(ctx, c, encoded, item.replayAfterSuccessfulDelivery); err != nil {
					return err
				}
			} else if err := s.replayRPCResultByRequest(ctx, c, item.msgID); err != nil {
				return err
			}
		case inboundItemRewrappedRPC:
			// Activation is deferred until every session/control barrier has
			// committed. It subscribes to the original result event and never waits
			// or dispatches the business handler again.
			continue
		case inboundItemPing:
			if err := s.sendPong(ctx, c, item.msgID, item.payload.(mt.PingRequest).PingID); err != nil {
				return err
			}
		case inboundItemPingDelay:
			if err := s.sendPong(ctx, c, item.msgID, item.payload.(mt.PingDelayDisconnectRequest).PingID); err != nil {
				return err
			}
		case inboundItemFutureSalts:
			if err := s.sendFutureSalts(ctx, c, item.msgID, item.payload.(mt.GetFutureSaltsRequest).Num); err != nil {
				return err
			}
		case inboundItemMsgsAck:
			ids := item.payload.(int64VectorView).materialize()
			c.AckServerMessages(ids)
			s.log.Debug("Received msgs_ack", zap.Int64s("msg_ids", ids))
		case inboundItemStateReq:
			ids := item.payload.(int64VectorView).materialize()
			outgoing, err := c.OutgoingStateInfo(ctx, ids)
			if err != nil {
				return err
			}
			if err := s.sendMsgsStateInfo(ctx, c, item.msgID, mergeStateInfo(outgoing, cs.stateInfo(ids))); err != nil {
				return err
			}
		case inboundItemResendReq:
			ids := item.payload.(int64VectorView).materialize()
			outgoing, err := c.ResendMessages(ctx, ids)
			if err != nil {
				return err
			}
			if err := s.sendMsgsStateInfo(ctx, c, item.msgID, mergeStateInfo(outgoing, cs.stateInfo(ids))); err != nil {
				return err
			}
		case inboundItemStateInfo:
			value := item.payload.(stateInfoPayload)
			s.log.Debug("Received msgs_state_info", zap.Int64("req_msg_id", value.reqMsgID), zap.Int("len", len(value.info)))
		case inboundItemAllInfo:
			value := item.payload.(allInfoPayload)
			s.log.Debug("Received msgs_all_info", zap.Int("msg_ids", value.count), zap.Int("len", len(value.info)))
		case inboundItemDestroySession:
			if err := s.sendDestroySession(ctx, c, item.payload.(mt.DestroySessionRequest).SessionID); err != nil {
				return err
			}
		case inboundItemHTTPWait:
			value := item.payload.(mt.HTTPWaitRequest)
			s.log.Debug("Received http_wait", zap.Int("max_delay", value.MaxDelay), zap.Int("wait_after", value.WaitAfter), zap.Int("max_wait", value.MaxWait))
		case inboundItemDropAnswer:
			value := item.payload.(mt.RPCDropAnswerRequest)
			s.log.Debug("Received rpc_drop_answer", zap.Int64("req_msg_id", value.ReqMsgID))
			if err := s.sendResult(ctx, c, item.msgID, &mt.RPCAnswerUnknown{}); err != nil {
				return err
			}
		case inboundItemDestroyAuthKey:
			s.log.Debug("Received destroy_auth_key", zap.String("auth_key_id", c.authKeyHex))
			if err := s.authKeys.Delete(ctx, c.authKeyID); err != nil {
				s.log.Warn("Delete auth key failed", zap.String("auth_key_id", c.authKeyHex), zap.Error(err))
				return c.SendRequiredControl(ctx, proto.MessageServerResponse, &destroyAuthKeyRPCResult{
					RequestMessageID: item.msgID,
					ResultTypeID:     destroyAuthKeyFailTypeID,
				})
			}
			if registry, ok := s.layerRPC.(LayerRPCSessionProfileRegistry); ok {
				registry.ForgetNegotiatedAuthKey(c.authKeyID)
			}
			// Fence every other active/claiming generation before acknowledging the
			// deletion. The exact requester remains writable only long enough to put the
			// request-correlated rpc_result(destroy_auth_key_ok) frame on the wire.
			s.conns.CloseSessionsForRawAuthKeyExceptConn(c.authKeyID, c)
			if err := c.SendRequiredControl(ctx, proto.MessageServerResponse, &destroyAuthKeyRPCResult{
				RequestMessageID: item.msgID,
				ResultTypeID:     destroyAuthKeyOkTypeID,
			}); err != nil {
				return err
			}
			// The key cannot reconnect to ACK or replay any old answer. Release every
			// logical outbox and receipt only after the terminal OK is physically on
			// the wire; retaining them for the offline TTL would be pure leakage.
			s.conns.ForgetLogicalSessionsForRawAuthKey(c.authKeyID)
			c.beginTerminalShutdown()
			c.closeTransport()
			return nil
		case inboundItemRPC:
			// prepareInboundRPCBatch owns every fresh RPC before synchronous service
			// execution begins; commitRPCBatch publishes them after all protocol barriers.
			continue
		case inboundItemCapacityError:
			if owner, _ := item.payload.(*rpcResultOwnerLease); owner != nil {
				owner.CompleteExecution(false)
			}
			if err := s.sendResult(ctx, c, item.msgID, rpcWorkerBusyError()); err != nil {
				return err
			}
		case inboundItemRPCAdmissionError:
			rpcErr, _ := item.payload.(*mt.RPCError)
			if rpcErr == nil {
				rpcErr = &mt.RPCError{ErrorCode: 400, ErrorMessage: "INPUT_REQUEST_INVALID"}
			}
			if err := s.sendResult(ctx, c, item.msgID, rpcErr); err != nil {
				return err
			}
		default:
			return fmt.Errorf("unknown inbound item kind %d", item.kind)
		}
	}
	return nil
}