package xturbo

import (
	"encoding/binary"
	"errors"
	"io"
	"os"

	"golang.org/x/sys/unix"
)

// MaxPacket bounds one server packet (a reply or an XGE event) in bytes. The
// read ring grows up to this size; a larger packet fails Dispatch with
// ErrPacketTooLarge and the connection is dead.
const MaxPacket = 16 << 20

// ErrPacketTooLarge reports a server packet longer than MaxPacket.
var ErrPacketTooLarge = errors.New("xturbo: packet too large")

// errReentrantDispatch is returned by Dispatch, and by the blocking calls
// built on it (Sync, InternAtoms, a NewXID refill), when called from a Handler
// callback. It is unexported: it reports a programming error, not a condition
// callers handle. The outer Dispatch is unaffected.
var errReentrantDispatch = errors.New("xturbo: Dispatch called from a Handler callback")

const (
	readBufSize    = 64 << 10
	packetHeadSize = 32
)

// Handler receives decoded packets from Dispatch, in wire order, on the
// goroutine calling Dispatch. OnReply sees only replies registered with
// Expect; OnError sees every error (Kind 0 when unregistered). A nil field
// drops that packet class (nil OnError counts Stats.UnhandledErrors). Fill it
// once at init; the pointers passed to callbacks are owned by the Conn and are
// reused for the next packet, so callbacks must not retain them, nor the
// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
// call Expect but must not call Dispatch, Sync, InternAtoms, nor NewXID when
// it needs a refill: those return an error there and change nothing.
type Handler struct {
	OnEvent func(ev *Event)
	OnReply func(r *Reply)
	OnError func(e *XError)
}

// Reply is one reply packet to a request registered with Expect. Seq is
// widened; Kind and Args are the registration's. Raw is the whole reply,
// 32+4*length bytes, valid only until the callback returns.
type Reply struct {
	Seq  Seq
	Kind uint16
	Args [4]uint32
	Raw  []byte
}

var noHandler Handler

// Dispatch reads what the server has sent without blocking and calls h for
// every complete packet, in wire order. It returns nil once the socket would
// block or after a bounded amount of input (about one read ring, so a flooding
// server cannot starve the caller's event loop): call it again while the fd is
// still readable, and Flush requests queued by callbacks in between. It
// returns io.EOF when the server closed the connection and ErrPacketTooLarge
// on a packet beyond MaxPacket. Read errors are sticky and also fail the Buf.
// Dispatch never flushes. A nil h drops every packet; a non-nil h is
// remembered for the packets Sync(nil), InternAtoms and NewXID read while they
// block. Errors on the requests root issues itself (reserved kinds) are
// handled internally and never reach OnError.
func (c *Conn) Dispatch(h *Handler) error {
	if c.dispatching {
		return errReentrantDispatch
	}
	c.dispatching = true
	defer func() { c.dispatching = false }()
	if c.fd < 0 {
		return ErrClosed
	}
	if c.rerr != nil {
		return c.rerr
	}
	if h == nil {
		h = &noHandler
	} else {
		c.h = h
	}
	budget := len(c.rb)
	c.rdIdle = false
	read := false
	for {
		if err := c.drain(h); err != nil {
			return c.failRead(err)
		}
		if c.fd < 0 { // a callback closed the Conn
			return ErrClosed
		}
		if budget <= 0 {
			return nil
		}
		n, err := unix.Read(c.fd, c.rb[c.rend:])
		switch {
		case err == unix.EINTR:
			continue
		case err == unix.EAGAIN:
			c.rdIdle = !read
			return nil
		case err == unix.ECONNRESET:
			return c.failRead(io.EOF)
		case err != nil:
			return c.failRead(os.NewSyscallError("read", err))
		case n == 0:
			return c.failRead(io.EOF)
		}
		c.rend += n
		budget -= n
		read = true
	}
}

func (c *Conn) failRead(err error) error {
	c.rerr = err
	if err == io.EOF {
		c.buf.Fail(ErrClosed)
	} else {
		c.buf.Fail(err)
	}
	return err
}

// drain handles every complete packet in the ring, then makes room for the
// next one (compacting, growing up to maxPacket).
func (c *Conn) drain(h *Handler) error {
	for c.fd >= 0 {
		avail := c.rend - c.rpos
		need := packetHeadSize
		if avail >= packetHeadSize {
			size, ok := packetSize(c.rb[c.rpos:], c.maxPacket)
			if !ok {
				return ErrPacketTooLarge
			}
			need = size
			if avail >= need {
				end := c.rpos + need
				p := c.rb[c.rpos:end:end]
				c.packet(h, p)
				if poison {
					fillPoison(p)
				}
				c.rpos = end
				continue
			}
		}
		c.makeRoom(need)
		break
	}
	return nil
}

// packetSize returns the length of the packet starting at p (len(p) >= 32):
// 32 for errors and events, 32+4*length for replies and XGE events.
func packetSize(p []byte, limit int) (int, bool) {
	n := uint64(packetHeadSize)
	if p[0] == 1 || p[0]&0x7f == codeGenericEvent {
		n += 4 * uint64(binary.LittleEndian.Uint32(p[4:]))
	}
	if n > uint64(limit) {
		return 0, false
	}
	return int(n), true
}

// makeRoom moves the unconsumed bytes to the start of the ring and grows it
// by doubling until need bytes fit. need <= maxPacket.
func (c *Conn) makeRoom(need int) {
	if c.rpos > 0 {
		k := copy(c.rb, c.rb[c.rpos:c.rend])
		if poison {
			fillPoison(c.rb[k:c.rend])
		}
		c.rpos, c.rend = 0, k
	}
	if need <= len(c.rb) {
		return
	}
	size := max(len(c.rb), packetHeadSize)
	for size < need {
		size *= 2
	}
	grown := make([]byte, min(size, c.maxPacket))
	copy(grown, c.rb[:c.rend])
	if poison {
		fillPoison(c.rb)
	}
	c.rb = grown
}

// packet decodes one complete packet and calls h.
func (c *Conn) packet(h *Handler, p []byte) {
	switch p[0] {
	case 0:
		decodeError(p, &c.ext, &c.xerr)
		c.xerr.Seq = c.widen(c.xerr.Seq)
		c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
		switch {
		case c.xerr.Kind != kindNone && c.xerr.Kind < kindUser:
			if !c.consumeError(&c.xerr) && h.OnError != nil {
				h.OnError(&c.xerr)
			}
		case h.OnError != nil:
			h.OnError(&c.xerr)
		default:
			c.stats.UnhandledErrors++
		}
	case 1:
		seq := c.widen(Seq(binary.LittleEndian.Uint16(p[2:])))
		kind, args, ok := c.take(seq)
		c.reply = Reply{Seq: seq, Kind: kind, Args: args, Raw: p}
		switch {
		case !ok:
			c.stats.DroppedReplies++
		case kind < kindUser:
			c.consumeReply(&c.reply)
		case h.OnReply != nil:
			h.OnReply(&c.reply)
		}
		c.reply.Raw = nil // do not retain the ring
	default:
		ev, _, ok := DecodeEvent(p, &c.ext)
		if !ok { // a sent event claiming code 0 or 1
			ev = Event{Kind: KindUnknown, SendEvent: true, Seq: Seq(binary.LittleEndian.Uint16(p[2:]))}
			copy(ev.Raw[:], p)
		}
		if p[0]&0x7f == codeKeymapNotify {
			ev.Seq = c.seq
		} else {
			ev.Seq = c.widen(ev.Seq)
		}
		if ev.Kind == KindUnknown {
			c.stats.UnknownEvents++
		}
		c.ev = ev
		if h.OnEvent != nil {
			h.OnEvent(&c.ev)
		}
		c.ev.Tail = nil // do not retain the ring
	}
}

// widen turns a 16-bit wire sequence into the full Seq, relative to the last
// one seen, acknowledges it to the Buf and retires the Expect slots of the
// requests it proves complete.
func (c *Conn) widen(wire Seq) Seq {
	prev := c.seq
	full := prev&^0xffff | wire&0xffff
	if full < prev {
		full += 0x10000
	}
	c.seq = full
	c.buf.ack(full)
	c.retire(prev, full)
	return full
}