package tunnel

import (
	"errors"
	"fmt"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"git.smesh.lol/nostrino/pkg/crypto/secp256k1"
	"git.smesh.lol/nostrino/pkg/nip44"
	"git.smesh.lol/nostrino/pkg/nostr"
	"git.smesh.lol/nostrino/pkg/relay"
)

type outFrame struct {
	data  []byte
	sent  time.Time
	tries int
}

type peerClosedError struct{ reason string }

func (e peerClosedError) Error() string {
	if e.reason == "" {
		return "tunnel: closed by peer"
	}
	return "tunnel: closed by peer: " + e.reason
}

type peerResetError struct{ reason string }

func (e peerResetError) Error() string {
	if e.reason == "" {
		return "tunnel: reset by peer"
	}
	return "tunnel: reset by peer: " + e.reason
}

type session struct {
	side     side
	relayMu  sync.Mutex
	relays   []*relay.Client
	key      [32]byte
	localPub [32]byte
	peerPub  [32]byte
	conv     [32]byte
	id       string
	chain    string

	window    int
	heartbeat time.Duration
	idle      time.Duration
	logf      func(string, ...any)

	// wireMu serialises event construction and relay writes so that one
	// session's frames do not interleave mid-frame at the transport.
	wireMu sync.Mutex

	// stateMu guards the sequence state and the send window.
	stateMu    sync.Mutex
	sendSeq    uint32
	unacked    map[uint32]*outFrame
	inflight   int
	peerAck    uint32
	sendCredit int
	ackCh      chan struct{}

	// receive state
	recvNext uint32
	recvBuf  map[uint32][]byte
	gapAt    time.Time

	mtuLen int64 // negotiated plaintext frame size, atomic

	inbox   chan *envelope
	writeCh chan []byte

	// held is the number of received frames the application has not consumed yet: the
	// frames parked in recvBuf waiting for a gap to fill, plus the frames in readQueue. It
	// is an atomic rather than a count under readMu because credit() must see both, and a
	// second lock taken while readMu is held could invert with pushRead.
	held int32

	// read queue
	readMu    sync.Mutex
	readQueue [][]byte
	readBytes int
	readWake  chan struct{}
	ackWake   chan struct{}

	// deadlines
	dlMu          sync.Mutex
	readDeadline  time.Time
	writeDeadline time.Time
	readDLWake    chan struct{}
	writeDLWake   chan struct{}

	lastRecv int64 // unix nanos, atomic
	lastSend int64 // unix nanos, atomic

	done      chan struct{}
	closeOnce sync.Once
	errMu     sync.Mutex
	err       error

	established chan struct{}
	estOnce     sync.Once

	subs    []*relay.Subscription
	onClose func()
}

func newSession(sd side, key [32]byte, localPub, peerPub [32]byte, id string, cfg DialConfig) (*session, error) {
	if !secp256k1.ValidatePubKey(peerPub) {
		return nil, errors.New("tunnel: peer key is not on the curve")
	}
	conv, err := nip44.ConversationKey(key, peerPub)
	if err != nil {
		return nil, fmt.Errorf("tunnel: conversation key: %w", err)
	}
	mtu := cfg.MTU
	if mtu == 0 {
		mtu = defaultMTU
	}
	window := cfg.Window
	if window == 0 {
		window = defaultWindow
	}
	heartbeat := cfg.Heartbeat
	if heartbeat == 0 {
		heartbeat = defaultHeartbeat
	}
	idle := cfg.IdleTimeout
	if idle == 0 {
		idle = defaultIdle
	}
	logf := cfg.Logf
	if logf == nil {
		logf = func(string, ...any) {}
	}
	s := &session{
		side:        sd,
		key:         key,
		localPub:    localPub,
		peerPub:     peerPub,
		conv:        conv,
		id:          id,
		chain:       cfg.Chain,
		window:      window,
		heartbeat:   heartbeat,
		idle:        idle,
		logf:        logf,
		unacked:     make(map[uint32]*outFrame),
		sendCredit:  window,
		recvNext:    1,
		recvBuf:     make(map[uint32][]byte),
		ackCh:       make(chan struct{}, 1),
		inbox:       make(chan *envelope, inboxFrames),
		writeCh:     make(chan []byte, sendQueueFrames),
		readWake:    make(chan struct{}, 1),
		ackWake:     make(chan struct{}, 1),
		readDLWake:  make(chan struct{}),
		writeDLWake: make(chan struct{}),
		done:        make(chan struct{}),
		established: make(chan struct{}),
	}
	atomic.StoreInt64(&s.mtuLen, int64(mtu))
	atomic.StoreInt64(&s.lastRecv, time.Now().UnixNano())
	return s, nil
}

func (s *session) mtu() int { return int(atomic.LoadInt64(&s.mtuLen)) }

func (s *session) start() {
	go s.sendLoop()
	go s.recvLoop()
	go s.ackLoop()
	if s.side == sideClient {
		for _, sub := range s.subs {
			sub := sub
			go s.clientReadLoop(sub)
		}
	}
}

// setRelays installs the relays this session publishes on. The client owns all
// of them; the gateway shares them with its listener.
func (s *session) setRelays(relays []*relay.Client, subs []*relay.Subscription) {
	s.relayMu.Lock()
	s.relays = append([]*relay.Client(nil), relays...)
	s.subs = append([]*relay.Subscription(nil), subs...)
	s.relayMu.Unlock()
}

func (s *session) addRelay(rc *relay.Client) bool {
	s.relayMu.Lock()
	defer s.relayMu.Unlock()
	for _, existing := range s.relays {
		if existing == rc {
			return false
		}
	}
	s.relays = append(s.relays, rc)
	return true
}

func (s *session) removeRelay(rc *relay.Client) {
	s.relayMu.Lock()
	defer s.relayMu.Unlock()
	for i, existing := range s.relays {
		if existing == rc {
			s.relays = append(s.relays[:i], s.relays[i+1:]...)
			return
		}
	}
}

func (s *session) relayList() []*relay.Client {
	s.relayMu.Lock()
	defer s.relayMu.Unlock()
	return append([]*relay.Client(nil), s.relays...)
}

func (s *session) localKind() int {
	if s.side == sideGateway {
		return KindGateway
	}
	return KindClient
}

func (s *session) buildEvent(env *envelope) (*nostr.Event, error) {
	raw, err := env.marshal()
	if err != nil {
		return nil, err
	}
	content, err := nip44.Encrypt(s.conv, raw)
	if err != nil {
		return nil, fmt.Errorf("tunnel: encrypt: %w", err)
	}
	return &nostr.Event{
		Kind:      s.localKind(),
		CreatedAt: time.Now().Unix(),
		Tags: nostr.Tags{
			{"p", nostr.Hex(s.peerPub[:])},
			{"s", s.id},
			{"enc", "nip44-v2"},
			{"v", "1"},
		},
		Content: content,
	}, nil
}

func (s *session) sendEnvelope(env *envelope) error {
	ev, err := s.buildEvent(env)
	if err != nil {
		return err
	}
	s.wireMu.Lock()
	defer s.wireMu.Unlock()
	relays := s.relayList()
	if len(relays) == 0 {
		return errors.New("tunnel: no relay")
	}
	sent := 0
	var lastErr error
	for _, rc := range relays {
		if err := rc.Send(ev); err != nil {
			lastErr = err
			s.removeRelay(rc)
			s.logf("tunnel: relay send failed: %v", err)
			continue
		}
		sent++
	}
	if sent == 0 {
		return fmt.Errorf("tunnel: relay send: %w", lastErr)
	}
	atomic.StoreInt64(&s.lastSend, time.Now().UnixNano())
	return nil
}

func (s *session) sendHelloAck() {
	env := &envelope{
		V: protocolVersion, S: s.id, T: frameHelloAck,
		MTU: s.mtu(), Win: s.window, PV: protocolVersion, Chain: s.chain,
	}
	if err := s.sendEnvelope(env); err != nil {
		s.closeWithErr(err)
	}
}

func (s *session) handleHelloAck(env *envelope) error {
	if env.PV != protocolVersion {
		return fmt.Errorf("tunnel: gateway protocol version %d", env.PV)
	}
	if env.Chain != "" && s.chain != "" && env.Chain != s.chain {
		return fmt.Errorf("tunnel: gateway chain %q, want %q", env.Chain, s.chain)
	}
	if env.MTU >= minMTU && env.MTU < s.mtu() {
		atomic.StoreInt64(&s.mtuLen, int64(env.MTU))
	}
	if env.Win >= minWindow && env.Win < s.window {
		s.window = env.Win
	}
	s.stateMu.Lock()
	if s.sendCredit > s.window {
		s.sendCredit = s.window
	}
	s.stateMu.Unlock()
	if env.Chain != "" {
		s.chain = env.Chain
	}
	s.estOnce.Do(func() { close(s.established) })
	return nil
}

func (s *session) clientReadLoop(sub *relay.Subscription) {
	for {
		select {
		case <-s.done:
			return
		case <-sub.Done():
			// One relay may fail while another still carries the stream. Only
			// close when the session has no subscriptions left.
			s.relayMu.Lock()
			remaining := 0
			for _, other := range s.subs {
				if other != sub {
					select {
					case <-other.Done():
					default:
						remaining++
					}
				}
			}
			s.relayMu.Unlock()
			if remaining == 0 {
				s.closeWithErr(errors.New("tunnel: all relay subscriptions ended"))
			}
			return
		case ev := <-sub.Events:
			if ev == nil || ev.Kind != KindGateway || ev.PubKey != s.peerPub {
				continue
			}
			raw, err := nip44.Decrypt(s.conv, ev.Content)
			if err != nil {
				continue
			}
			env, err := parseEnvelope(raw)
			if err != nil || env.S != s.id {
				continue
			}
			s.enqueue(env)
		}
	}
}

func (s *session) enqueue(env *envelope) {
	select {
	case s.inbox <- env:
	default:
		// The receiver is behind. Dropping a data frame is safe: the gap
		// triggers a NACK and a retransmission. Dropping a control frame is
		// recovered by the heartbeat.
	}
}

func (s *session) recvLoop() {
	for {
		select {
		case <-s.done:
			return
		case env := <-s.inbox:
			atomic.StoreInt64(&s.lastRecv, time.Now().UnixNano())
			var err error
			switch env.T {
			case frameData:
				err = s.handleData(env)
			case frameAck:
				s.handleAck(env)
			case frameNack:
				err = s.handleNack(env.Nack)
			case framePing:
				err = s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: framePong, Nonce: env.Nonce})
			case framePong:
			case frameHelloAck:
				if s.side == sideClient {
					err = s.handleHelloAck(env)
				}
			case frameHello:
				// Duplicate HELLO; the gateway answered the original.
			case frameClose:
				s.closeWithErr(peerClosedError{env.Reason})
				return
			case frameReset:
				s.resetWithErr(peerResetError{env.Reason})
				return
			default:
				err = fmt.Errorf("tunnel: unknown frame type %q", env.T)
			}
			if err != nil {
				s.closeWithErr(err)
				return
			}
		}
	}
}

func (s *session) handleData(env *envelope) error {
	if env.Seq == 0 {
		return errors.New("tunnel: data frame with sequence zero")
	}
	var deliver [][]byte
	var ack uint32
	var sendAck bool
	s.stateMu.Lock()
	switch {
	case env.Seq == s.recvNext:
		deliver = append(deliver, env.Data)
		// This frame arrived in order and was never buffered, so it is newly held.
		atomic.AddInt32(&s.held, 1)
		s.recvNext++
		for {
			d, ok := s.recvBuf[s.recvNext]
			if !ok {
				break
			}
			// Drained frames were counted when they were buffered.
			delete(s.recvBuf, s.recvNext)
			deliver = append(deliver, d)
			s.recvNext++
		}
		s.gapAt = time.Time{}
		ack = s.recvNext - 1
		sendAck = true
	case seqBefore(env.Seq, s.recvNext):
		ack = s.recvNext - 1
		sendAck = true
	default:
		// The defensive bound: with correct credit a sender cannot have more than
		// window frames unacknowledged, so no frame can be a whole window ahead of
		// what this side expects. This used to allow window*2, which papered over a
		// credit bug and then failed at 2*window+1.
		if env.Seq-s.recvNext >= uint32(s.window) {
			ahead := env.Seq - s.recvNext
			s.stateMu.Unlock()
			return fmt.Errorf(
				"%w: frame %d is %d ahead of %d, the window is %d",
				ErrFrameAhead, env.Seq, ahead, s.recvNext, s.window,
			)
		}
		if _, exists := s.recvBuf[env.Seq]; !exists {
			s.recvBuf[env.Seq] = env.Data
			atomic.AddInt32(&s.held, 1)
		}
		now := time.Now()
		nack := false
		if s.gapAt.IsZero() {
			s.gapAt = now
		} else if now.Sub(s.gapAt) >= nackDelay {
			s.gapAt = now
			nack = true
		}
		missing := missingSeqs(s.recvNext, env.Seq, 32)
		s.stateMu.Unlock()
		if nack {
			return s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameNack, Nack: missing})
		}
		return nil
	}
	s.stateMu.Unlock()

	for _, d := range deliver {
		s.pushRead(d)
	}
	if sendAck {
		return s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameAck, Ack: ack, Win: int(s.credit())})
	}
	return nil
}

// pushRead queues delivered bytes for Read.
//
// The frame is counted as held by handleData when it arrives — once, whether it arrived in
// order or was parked in recvBuf and drained after a gap filled. read takes the count back
// down when the application consumes it. The count is per frame, not per byte, because credit
// is a frame budget.
func (s *session) pushRead(data []byte) {
	s.readMu.Lock()
	s.readQueue = append(s.readQueue, data)
	s.readBytes += len(data)
	s.readMu.Unlock()
	select {
	case s.readWake <- struct{}{}:
	default:
	}
}

// credit is the number of frames the application has room for. It is carried
// in every acknowledgement; the sender may not send new data without credit.
// A frozen worker therefore stops the bridge's send side and the node's TCP
// window takes over, rather than the browser queuing filters without bound.
//
// It counts every frame the application has not consumed, not just the ones
// already in readQueue: a frame held in recvBuf waiting for a gap has arrived
// and occupies application room just the same, and a receiver that forgot it
// would advertise room for frames it is already holding.
func (s *session) credit() uint32 {
	free := s.window - int(atomic.LoadInt32(&s.held))
	if free < 0 {
		free = 0
	}
	if free > s.window {
		free = s.window
	}
	return uint32(free)
}

func (s *session) ackLoop() {
	for {
		select {
		case <-s.done:
			return
		case <-s.ackWake:
			s.stateMu.Lock()
			ack := s.recvNext - 1
			s.stateMu.Unlock()
			if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameAck, Ack: ack, Win: int(s.credit())}); err != nil {
				s.closeWithErr(err)
				return
			}
		}
	}
}

func (s *session) handleAck(env *envelope) {
	ack := env.Ack
	if ack == 0 {
		return
	}
	s.stateMu.Lock()
	if seqAfter(ack, s.sendSeq) {
		s.stateMu.Unlock()
		s.closeWithErr(fmt.Errorf("tunnel: peer acknowledged unsent frame %d", ack))
		return
	}
	newest := seqAfter(ack, s.peerAck)
	if newest {
		s.peerAck = ack
		for seq := range s.unacked {
			if !seqAfter(seq, ack) {
				delete(s.unacked, seq)
				s.inflight--
			}
		}
		if s.inflight < 0 {
			s.inflight = 0
		}
	}
	// The receiver's free-frame count is authoritative. A stale acknowledgement
	// must not take credit away, so only the newest cumulative ack applies it.
	if newest || ack == s.peerAck {
		// The peer's Win is the room its application has, but this sender's
		// unacknowledged frames are still in the pipe and still occupy that room. Credit
		// is therefore the smaller of what the peer advertises and what this side may
		// still have outstanding; taking Win alone let a sender push a second window
		// while the first was unanswered.
		free := s.window - s.inflight
		if free < 0 {
			free = 0
		}
		wnd := int(env.Win)
		if wnd > free {
			wnd = free
		}
		if wnd < 0 {
			wnd = 0
		}
		s.sendCredit = wnd
	}
	s.stateMu.Unlock()
	select {
	case s.ackCh <- struct{}{}:
	default:
	}
}

func (s *session) handleNack(seqs []uint32) error {
	type resend struct {
		seq  uint32
		data []byte
	}
	var out []resend
	s.stateMu.Lock()
	for _, seq := range seqs {
		if f := s.unacked[seq]; f != nil {
			f.tries++
			f.sent = time.Now()
			out = append(out, resend{seq: seq, data: f.data})
		}
	}
	s.stateMu.Unlock()
	for _, r := range out {
		ack := s.currentAck()
		if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameData, Seq: r.seq, Ack: ack, Win: int(s.credit()), Data: r.data}); err != nil {
			return err
		}
	}
	return nil
}

func (s *session) currentAck() uint32 {
	s.stateMu.Lock()
	defer s.stateMu.Unlock()
	if s.recvNext == 0 {
		return 0
	}
	return s.recvNext - 1
}

func (s *session) sendLoop() {
	ticker := time.NewTicker(250 * time.Millisecond)
	defer ticker.Stop()
	for {
		select {
		case <-s.done:
			return
		case chunk := <-s.writeCh:
			if err := s.sendData(chunk); err != nil {
				s.closeWithErr(err)
				return
			}
		case <-s.ackCh:
		case now := <-ticker.C:
			if err := s.tick(now); err != nil {
				s.closeWithErr(err)
				return
			}
		}
	}
}

func (s *session) sendData(chunk []byte) error {
	if err := s.acquireCredit(); err != nil {
		return err
	}
	s.stateMu.Lock()
	s.sendSeq++
	seq := s.sendSeq
	s.unacked[seq] = &outFrame{data: chunk, sent: time.Now(), tries: 1}
	s.inflight++
	ack := uint32(0)
	if s.recvNext > 0 {
		ack = s.recvNext - 1
	}
	s.stateMu.Unlock()
	if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameData, Seq: seq, Ack: ack, Win: int(s.credit()), Data: chunk}); err != nil {
		s.stateMu.Lock()
		delete(s.unacked, seq)
		s.inflight--
		s.sendCredit++
		s.stateMu.Unlock()
		return err
	}
	return nil
}

// acquireCredit takes one frame of receive credit, waiting for the peer to
// acknowledge and advertise room. A sender never has more than the negotiated
// window of unacknowledged data in the relay.
func (s *session) acquireCredit() error {
	for {
		s.stateMu.Lock()
		if s.sendCredit > 0 {
			s.sendCredit--
			s.stateMu.Unlock()
			return nil
		}
		s.stateMu.Unlock()
		select {
		case <-s.ackCh:
		case <-s.done:
			return s.Err()
		}
	}
}

func (s *session) tick(now time.Time) error {
	// Idle check: a stream that has heard nothing for the idle bound is dead
	// even if the local write side is still buffered.
	lastRecv := time.Unix(0, atomic.LoadInt64(&s.lastRecv))
	if s.idle > 0 && now.Sub(lastRecv) > s.idle {
		return fmt.Errorf("tunnel: idle for %s: %w", now.Sub(lastRecv).Round(time.Second), ErrTimeout)
	}
	// Reap frames the peer has not acknowledged.
	var resend []struct {
		seq  uint32
		data []byte
	}
	dead := false
	s.stateMu.Lock()
	for seq, f := range s.unacked {
		if now.Sub(f.sent) < retransmitAfter {
			continue
		}
		if f.tries >= maxFrameRetries {
			dead = true
			break
		}
		f.tries++
		f.sent = now
		resend = append(resend, struct {
			seq  uint32
			data []byte
		}{seq, f.data})
	}
	s.stateMu.Unlock()
	if dead {
		return errors.New("tunnel: frame retry limit reached")
	}
	sort.Slice(resend, func(i, j int) bool { return seqBefore(resend[i].seq, resend[j].seq) })
	for _, r := range resend {
		if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameData, Seq: r.seq, Ack: s.currentAck(), Win: int(s.credit()), Data: r.data}); err != nil {
			return err
		}
	}
	lastSend := time.Unix(0, atomic.LoadInt64(&s.lastSend))
	if s.heartbeat > 0 && now.Sub(lastSend) >= s.heartbeat {
		return s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: framePing, Nonce: uint64(now.UnixNano())})
	}
	return nil
}

func (s *session) write(p []byte) (int, error) {
	total := 0
	for total < len(p) {
		n := len(p) - total
		if mtu := s.mtu(); n > mtu {
			n = mtu
		}
		chunk := make([]byte, n)
		copy(chunk, p[total:total+n])
		if err := s.queue(chunk); err != nil {
			return total, err
		}
		total += n
	}
	return total, nil
}

func (s *session) queue(chunk []byte) error {
	for {
		select {
		case s.writeCh <- chunk:
			return nil
		case <-s.done:
			return s.Err()
		default:
		}
		dl := s.getWriteDeadline()
		var timer *time.Timer
		var tc <-chan time.Time
		if !dl.IsZero() {
			remaining := time.Until(dl)
			if remaining <= 0 {
				return ErrTimeout
			}
			timer = time.NewTimer(remaining)
			tc = timer.C
		}
		wake := s.getWriteWake()
		select {
		case s.writeCh <- chunk:
			stopTimer(timer)
			return nil
		case <-s.done:
			stopTimer(timer)
			return s.Err()
		case <-wake:
			stopTimer(timer)
			continue
		case <-tc:
			return ErrTimeout
		}
	}
}

func stopTimer(t *time.Timer) {
	if t != nil {
		t.Stop()
	}
}

func (s *session) read(p []byte) (int, error) {
	if len(p) == 0 {
		return 0, nil
	}
	for {
		s.readMu.Lock()
		if len(s.readQueue) > 0 {
			chunk := s.readQueue[0]
			n := copy(p, chunk)
			if n < len(chunk) {
				s.readQueue[0] = chunk[n:]
			} else {
				s.readQueue[0] = nil
				s.readQueue = s.readQueue[1:]
			}
			s.readBytes -= n
			consumed := n == len(chunk)
			s.readMu.Unlock()
			if consumed {
				// The application has taken this whole frame, so its room comes back.
				atomic.AddInt32(&s.held, -1)
				select {
				case s.ackWake <- struct{}{}:
				default:
				}
			}
			return n, nil
		}
		s.readMu.Unlock()

		dl := s.getReadDeadline()
		var timer *time.Timer
		var tc <-chan time.Time
		if !dl.IsZero() {
			remaining := time.Until(dl)
			if remaining <= 0 {
				return 0, ErrTimeout
			}
			timer = time.NewTimer(remaining)
			tc = timer.C
		}
		wake := s.getReadWake()
		select {
		case <-s.readWake:
			stopTimer(timer)
		case <-s.done:
			stopTimer(timer)
			return 0, s.Err()
		case <-wake:
			stopTimer(timer)
		case <-tc:
			return 0, ErrTimeout
		}
	}
}

func (s *session) setReadDeadline(t time.Time) {
	s.dlMu.Lock()
	s.readDeadline = t
	close(s.readDLWake)
	s.readDLWake = make(chan struct{})
	s.dlMu.Unlock()
}

func (s *session) setWriteDeadline(t time.Time) {
	s.dlMu.Lock()
	s.writeDeadline = t
	close(s.writeDLWake)
	s.writeDLWake = make(chan struct{})
	s.dlMu.Unlock()
}

func (s *session) getReadDeadline() time.Time {
	s.dlMu.Lock()
	defer s.dlMu.Unlock()
	return s.readDeadline
}

func (s *session) getWriteDeadline() time.Time {
	s.dlMu.Lock()
	defer s.dlMu.Unlock()
	return s.writeDeadline
}

func (s *session) getReadWake() <-chan struct{} {
	s.dlMu.Lock()
	defer s.dlMu.Unlock()
	return s.readDLWake
}

func (s *session) getWriteWake() <-chan struct{} {
	s.dlMu.Lock()
	defer s.dlMu.Unlock()
	return s.writeDLWake
}

// Err returns why the stream ended, or nil while it is open.
func (s *session) Err() error {
	s.errMu.Lock()
	defer s.errMu.Unlock()
	if s.err != nil {
		return s.err
	}
	return ErrClosed
}

func (s *session) closeWithErr(err error) { s.closeWith(err, frameClose) }

func (s *session) resetWithErr(err error) { s.closeWith(err, frameReset) }

// closeWith ends the session. A local Close sends CLOSE (orderly) or RESET
// (abort); a session already ended by the peer or by a transport error does
// not send another frame. RESET is the cancellation primitive: the caller is
// abandoning the stream, so no in-flight range matters any more.
func (s *session) closeWith(err error, frame string) {
	s.closeOnce.Do(func() {
		if err == nil {
			err = ErrClosed
		}
		s.errMu.Lock()
		s.err = err
		s.errMu.Unlock()

		var closed peerClosedError
		var reset peerResetError
		if !errors.As(err, &closed) && !errors.As(err, &reset) && !errors.Is(err, ErrClosed) {
			_ = s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frame, Reason: err.Error()})
		}
		close(s.done)
		s.relayMu.Lock()
		subs := append([]*relay.Subscription(nil), s.subs...)
		relays := append([]*relay.Client(nil), s.relays...)
		s.relayMu.Unlock()
		for _, sub := range subs {
			sub.Close()
		}
		if s.side == sideClient {
			for _, rc := range relays {
				rc.Close()
			}
		}
		if s.onClose != nil {
			s.onClose()
		}
	})
}

func seqBefore(a, b uint32) bool { return int32(a-b) < 0 }
func seqAfter(a, b uint32) bool  { return int32(a-b) > 0 }

func missingSeqs(from, to uint32, max int) []uint32 {
	var out []uint32
	for seq := from; seqBefore(seq, to) && len(out) < max; seq++ {
		out = append(out, seq)
	}
	return out
}