// Package chaindata is the wallet's channel to a light client: a binary frame protocol over
// a tiny-spawn duplex.
//
// It is the seam, and the seam is the whole design. The wallet sends a frame and gets a
// frame back; what sits behind the other end is nostrino's light client today, a full node or a
// fixture tomorrow, and neither side of the channel needs to know which. The vocabulary is
// the old client's own: status, sync, match, and a batched filter read.
//
// Three properties are load-bearing:
//
//   - Frames are binary, not JSON. A filter set and a script list are byte strings; encoding
//     them as base64 would grow the payload and make every message a parsing problem. The
//     codec is a fixed 16-byte header and a payload, and the payload of a filter read is one
//     buffer that the browser transport can transfer rather than copy.
//   - Every request carries an id and every answer echoes it. The wallet may have one
//     request in flight and a cancel behind it, so an answer with the wrong id is a bug
//     rather than an answer.
//   - A close is a frame, and it also means the far end died. A worker that is terminated
//     without a word closes the port; the domain treats that exactly like a close frame, so
//     an in-flight request fails instead of waiting for an answer that cannot come.
package chaindata

import (
	"encoding/binary"
	"errors"
	"fmt"
	"sort"
	"sync"
	"time"

	"git.smesh.lol/tiny-spawn"
)

// Frame layout constants. The header is fixed so a decoder can reject anything it does not
// understand before it allocates.
const (
	// Magic is 'N','C','D','1' read little-endian.
	Magic uint32 = 0x3144434e
	// Version is the frame format. A mismatch is refused rather than guessed at.
	Version uint8 = 1
	// HeaderLen is the fixed part of every frame:
	//
	//	0..4  magic   4 version   5 kind   6..8 flags
	//	8..12 id     12..14 method 14..16 status  16..20 payload length
	//
	// The length is in the header so a decoder rejects a truncated frame before it
	// allocates the payload.
	HeaderLen = 20
	// MaxPayload bounds one frame. A filter range is the largest thing that crosses this
	// channel, and a batch is a few hundred kilobytes.
	MaxPayload = 8 << 20
	// MaxScripts and MaxScriptLen bound a match request before anything is allocated.
	MaxScripts   = 10000
	MaxScriptLen = 10000
	// MaxFilters bounds one filter read.
	MaxFilters = 2000
)

// Kind is what a frame is for.
type Kind uint8

const (
	// KindRequest asks the far end to do something.
	KindRequest Kind = 0
	// KindResult answers a request.
	KindResult Kind = 1
	// KindCancel asks the far end to abandon one request.
	KindCancel Kind = 2
	// KindClose says this end is going away. The far end must finish or fail what it is
	// doing and stop.
	KindClose Kind = 3
)

// String renders a kind for a log or an error.
func (k Kind) String() string {
	switch k {
	case KindRequest:
		return "request"
	case KindResult:
		return "result"
	case KindCancel:
		return "cancel"
	case KindClose:
		return "close"
	}

	return fmt.Sprintf("kind(%d)", uint8(k))
}

// Method is one command a request may carry.
type Method uint16

const (
	// MethodStatus reports the chain position.
	MethodStatus Method = 1
	// MethodSync advances headers and filters and reports the new position.
	MethodSync Method = 2
	// MethodMatch returns the heights whose filters mention any of the scripts.
	MethodMatch Method = 3
	// MethodBlock returns one raw block, which is what turns a matched height into coins.
	MethodBlock Method = 5
	// MethodBroadcast sends one raw transaction to the peer the light client follows, so a
	// wallet that signs locally has somewhere to hand the result.
	MethodBroadcast Method = 6
	// MethodFilters returns a contiguous run of compact filters, one payload, so a wallet
	// that needs them fetches a batch rather than a round trip per block.
	MethodFilters Method = 4
)

// String renders a method.
func (m Method) String() string {
	switch m {
	case MethodStatus:
		return "status"
	case MethodSync:
		return "sync"
	case MethodMatch:
		return "match"
	case MethodBlock:
		return "block"
	case MethodBroadcast:
		return "broadcast"
	case MethodFilters:
		return "filters"
	}

	return fmt.Sprintf("method(%d)", uint16(m))
}

// Frame is one message.
//
// Payload is the only variable part. It is a slice rather than a string because it is
// usually not text: scripts, filters, heights and status words are binary.
type Frame struct {
	Kind Kind
	// ID correlates a result or a cancel with its request. Zero is not used for requests.
	ID uint32
	// Method is set on requests and on results, so an answer says what it answers even if
	// its id is all a strict caller checks.
	Method Method
	// Status is zero on success and a small code on failure. It is a code rather than a
	// string so a transport never has to parse an error.
	Status Status
	// Payload carries the method's arguments or its answer.
	Payload []byte
}

// Status is a result code.
type Status uint16

const (
	// StatusOK is a successful result.
	StatusOK Status = 0
	// StatusFailed is a refusal the source reported.
	StatusFailed Status = 1
	// StatusCanceled means the request was cancelled before it finished.
	StatusCanceled Status = 2
	// StatusUnknownMethod is an unsupported method.
	StatusUnknownMethod Status = 3
	// StatusMalformed means the payload could not be read.
	StatusMalformed Status = 4
	// StatusClosed means the far end went away.
	StatusClosed Status = 5
)

// String renders a status.
func (s Status) String() string {
	switch s {
	case StatusOK:
		return "ok"
	case StatusFailed:
		return "failed"
	case StatusCanceled:
		return "canceled"
	case StatusUnknownMethod:
		return "unknown method"
	case StatusMalformed:
		return "malformed"
	case StatusClosed:
		return "closed"
	}

	return fmt.Sprintf("status(%d)", uint16(s))
}

// ErrPayloadTooLarge is returned by Encode when a payload exceeds MaxPayload.
var ErrPayloadTooLarge = errors.New("chaindata: payload is larger than one frame")

// ErrBadFrame is returned by Decode for anything that is not a frame this version reads.
var ErrBadFrame = errors.New("chaindata: not a chaindata frame")

// Encode writes a frame.
//
// The header is written first and the payload copied once, so a batch of filters is one
// allocation and one pass rather than an encoder walking the items.
func Encode(frame Frame) ([]byte, error) {
	if len(frame.Payload) > MaxPayload {
		return nil, fmt.Errorf("%w: %d bytes", ErrPayloadTooLarge, len(frame.Payload))
	}

	out := make([]byte, HeaderLen+len(frame.Payload))

	binary.LittleEndian.PutUint32(out[0:4], Magic)
	out[4] = Version
	out[5] = byte(frame.Kind)
	binary.LittleEndian.PutUint16(out[6:8], 0)
	binary.LittleEndian.PutUint32(out[8:12], frame.ID)
	binary.LittleEndian.PutUint16(out[12:14], uint16(frame.Method))
	binary.LittleEndian.PutUint16(out[14:16], uint16(frame.Status))
	binary.LittleEndian.PutUint32(out[16:20], uint32(len(frame.Payload)))

	copy(out[HeaderLen:], frame.Payload)

	return out, nil
}

// Decode reads a frame and refuses anything that is not one.
func Decode(raw []byte) (Frame, error) {
	if len(raw) < HeaderLen {
		return Frame{}, fmt.Errorf("%w: %d bytes is shorter than a header", ErrBadFrame,
			len(raw))
	}

	if magic := binary.LittleEndian.Uint32(raw[0:4]); magic != Magic {
		return Frame{}, fmt.Errorf("%w: wrong magic", ErrBadFrame)
	}

	if version := raw[4]; version != Version {
		return Frame{}, fmt.Errorf("%w: version %d", ErrBadFrame, version)
	}

	kind := Kind(raw[5])

	if kind > KindClose {
		return Frame{}, fmt.Errorf("%w: %s", ErrBadFrame, kind)
	}

	length := binary.LittleEndian.Uint32(raw[16:20])

	if length > MaxPayload {
		return Frame{}, fmt.Errorf("%w: %d-byte payload", ErrBadFrame, length)
	}

	if uint64(len(raw)) != uint64(HeaderLen)+uint64(length) {
		return Frame{}, fmt.Errorf("%w: %d bytes of payload, not %d", ErrBadFrame,
			len(raw)-HeaderLen, length)
	}

	payload := make([]byte, length)
	copy(payload, raw[HeaderLen:])

	return Frame{
		Kind:    kind,
		ID:      binary.LittleEndian.Uint32(raw[8:12]),
		Method:  Method(binary.LittleEndian.Uint16(raw[12:14])),
		Status:  Status(binary.LittleEndian.Uint16(raw[14:16])),
		Payload: payload,
	}, nil
}

//
// Payload codecs. Each one reads and writes exactly one shape, with the count read before
// anything is allocated from it.
//

// Position is the chain position a status reports.
type Position struct {
	// Tip is the highest verified header height.
	Tip uint32
	// Filters is the highest height whose filter is downloaded and verified.
	Filters uint32
	// Running reports whether a sync is in progress.
	Running bool
}

// PutPosition writes a position: tip, filters, flags.
func PutPosition(buffer []byte, position Position) []byte {
	buffer = binary.LittleEndian.AppendUint32(buffer, position.Tip)
	buffer = binary.LittleEndian.AppendUint32(buffer, position.Filters)

	flags := byte(0)

	if position.Running {
		flags = 1
	}

	return append(buffer, flags)
}

// ReadPosition reads a position.
func ReadPosition(payload []byte) (Position, error) {
	if len(payload) < 9 {
		return Position{}, fmt.Errorf("%w: a position is 9 bytes", ErrBadFrame)
	}

	return Position{
		Tip:     binary.LittleEndian.Uint32(payload[0:4]),
		Filters: binary.LittleEndian.Uint32(payload[4:8]),
		Running: payload[8]&1 == 1,
	}, nil
}

// PutHeight writes the four-byte height a block request carries.
func PutHeight(buffer []byte, height uint32) []byte {
	return binary.LittleEndian.AppendUint32(buffer, height)
}

// ReadHeight reads a height, refusing anything that is not exactly four bytes.
func ReadHeight(payload []byte) (uint32, error) {
	if len(payload) != 4 {
		return 0, fmt.Errorf("%w: a height is 4 bytes, not %d", ErrBadFrame, len(payload))
	}

	return binary.LittleEndian.Uint32(payload), nil
}

// PutHeights writes a list of heights.
func PutHeights(buffer []byte, heights []uint32) []byte {
	buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(heights)))

	for _, height := range heights {
		buffer = binary.LittleEndian.AppendUint32(buffer, height)
	}

	return buffer
}

// ReadHeights reads a list of heights.
func ReadHeights(payload []byte) ([]uint32, error) {
	if len(payload) < 4 {
		return nil, fmt.Errorf("%w: a height list starts with a count", ErrBadFrame)
	}

	count := binary.LittleEndian.Uint32(payload[0:4])

	if count > MaxFilters*8 {
		return nil, fmt.Errorf("%w: %d heights", ErrBadFrame, count)
	}

	if uint64(len(payload)) != 4+uint64(count)*4 {
		return nil, fmt.Errorf("%w: %d heights need %d bytes, not %d", ErrBadFrame,
			count, 4+count*4, len(payload))
	}

	out := make([]uint32, count)

	for i := uint32(0); i < count; i++ {
		out[i] = binary.LittleEndian.Uint32(payload[4+i*4:])
	}

	return out, nil
}

// PutScripts writes a match request: a count, then each length and its bytes.
func PutScripts(buffer []byte, scripts [][]byte) ([]byte, error) {
	if len(scripts) == 0 {
		return nil, fmt.Errorf("a match needs at least one script")
	}

	if len(scripts) > MaxScripts {
		return nil, fmt.Errorf("%d scripts is more than one match", len(scripts))
	}

	buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(scripts)))

	for _, script := range scripts {
		if len(script) == 0 || len(script) > MaxScriptLen {
			return nil, fmt.Errorf("a script is 1 to %d bytes, not %d", MaxScriptLen,
				len(script))
		}

		buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(script)))
		buffer = append(buffer, script...)
	}

	return buffer, nil
}

// ReadScripts reads a match request.
func ReadScripts(payload []byte) ([][]byte, error) {
	if len(payload) < 4 {
		return nil, fmt.Errorf("%w: a match starts with a count", ErrBadFrame)
	}

	count := binary.LittleEndian.Uint32(payload[0:4])

	if count == 0 || count > MaxScripts {
		return nil, fmt.Errorf("%w: %d scripts", ErrBadFrame, count)
	}

	out := make([][]byte, 0, count)
	offset := 4

	for i := uint32(0); i < count; i++ {
		if offset+4 > len(payload) {
			return nil, fmt.Errorf("%w: script %d has no length", ErrBadFrame, i)
		}

		length := binary.LittleEndian.Uint32(payload[offset:])
		offset += 4

		if length == 0 || length > MaxScriptLen {
			return nil, fmt.Errorf("%w: script %d is %d bytes", ErrBadFrame, i, length)
		}

		if offset+int(length) > len(payload) {
			return nil, fmt.Errorf("%w: script %d runs past the payload", ErrBadFrame, i)
		}

		script := make([]byte, length)
		copy(script, payload[offset:offset+int(length)])

		out = append(out, script)
		offset += int(length)
	}

	if offset != len(payload) {
		return nil, fmt.Errorf("%w: %d trailing bytes", ErrBadFrame, len(payload)-offset)
	}

	return out, nil
}

// PutFilterRange writes a filter read request: from, count.
func PutFilterRange(buffer []byte, from, count uint32) []byte {
	buffer = binary.LittleEndian.AppendUint32(buffer, from)

	return binary.LittleEndian.AppendUint32(buffer, count)
}

// ReadFilterRange reads a filter read request.
func ReadFilterRange(payload []byte) (from, count uint32, err error) {
	if len(payload) != 8 {
		return 0, 0, fmt.Errorf("%w: a filter range is 8 bytes", ErrBadFrame)
	}

	from = binary.LittleEndian.Uint32(payload[0:4])
	count = binary.LittleEndian.Uint32(payload[4:8])

	if count == 0 || count > MaxFilters {
		return 0, 0, fmt.Errorf("%w: %d filters", ErrBadFrame, count)
	}

	return from, count, nil
}

// Filter is one compact filter and the block it belongs to.
type Filter struct {
	// Height is the block height.
	Height uint32
	// BlockHash is the block's hash, 32 bytes.
	BlockHash []byte
	// Filter is the Golomb-coded set.
	Filter []byte
}

// PutFilters writes a batch: a count, then each height, hash and filter.
//
// The whole batch is one payload, which is what the transport transfers instead of copying
// per filter.
func PutFilters(buffer []byte, filters []Filter) ([]byte, error) {
	if len(filters) > MaxFilters {
		return nil, fmt.Errorf("%d filters is more than one read", len(filters))
	}

	buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(filters)))

	for i, filter := range filters {
		if len(filter.BlockHash) != 32 {
			return nil, fmt.Errorf("filter %d has a %d-byte block hash", i,
				len(filter.BlockHash))
		}

		if len(filter.Filter) > MaxPayload {
			return nil, fmt.Errorf("filter %d is larger than one frame", i)
		}

		buffer = binary.LittleEndian.AppendUint32(buffer, filter.Height)
		buffer = append(buffer, filter.BlockHash...)
		buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(filter.Filter)))
		buffer = append(buffer, filter.Filter...)
	}

	return buffer, nil
}

// ReadFilters reads a batch of filters.
func ReadFilters(payload []byte) ([]Filter, error) {
	if len(payload) < 4 {
		return nil, fmt.Errorf("%w: a filter batch starts with a count", ErrBadFrame)
	}

	count := binary.LittleEndian.Uint32(payload[0:4])

	if count > MaxFilters {
		return nil, fmt.Errorf("%w: %d filters", ErrBadFrame, count)
	}

	out := make([]Filter, 0, count)
	offset := 4

	for i := uint32(0); i < count; i++ {
		if offset+4+32+4 > len(payload) {
			return nil, fmt.Errorf("%w: filter %d is truncated", ErrBadFrame, i)
		}

		height := binary.LittleEndian.Uint32(payload[offset:])
		offset += 4

		hash := make([]byte, 32)
		copy(hash, payload[offset:offset+32])
		offset += 32

		length := binary.LittleEndian.Uint32(payload[offset:])
		offset += 4

		if length > MaxPayload || offset+int(length) > len(payload) {
			return nil, fmt.Errorf("%w: filter %d is %d bytes", ErrBadFrame, i, length)
		}

		body := make([]byte, length)
		copy(body, payload[offset:offset+int(length)])

		offset += int(length)

		out = append(out, Filter{Height: height, BlockHash: hash, Filter: body})
	}

	if offset != len(payload) {
		return nil, fmt.Errorf("%w: %d trailing bytes", ErrBadFrame, len(payload)-offset)
	}

	sort.SliceStable(out, func(i, j int) bool { return out[i].Height < out[j].Height })

	return out, nil
}

//
// The domain.
//

// Source is the chain behind the channel: a light client, a full node or a fixture.
//
// Every method takes a context, because every one of them can be cancelled: a sync of ten
// thousand filters is the longest thing a wallet does, and a person closing the page should
// not have to wait for it.
type Source interface {
	// Position reports the current chain position.
	Position(ctx Context) (Position, error)
	// Sync advances the chain and reports where it ends.
	Sync(ctx Context, from uint32) (Position, error)
	// Match returns the ascending heights whose filters mention any of the scripts.
	Match(ctx Context, scripts [][]byte) ([]uint32, error)
	// Block returns the raw block at height.
	Block(ctx Context, height uint32) ([]byte, error)
	// Broadcast sends one raw transaction to the chain's peer.
	Broadcast(ctx Context, raw []byte) error
	// Filters returns a contiguous run of filters starting at from.
	Filters(ctx Context, from, count uint32) ([]Filter, error)
}

// Context is a cancellation handle for one request.
//
// It is an interface rather than context.Context so this package carries no dependency on
// the standard context's tree, and it carries a channel rather than a flag so a source can
// select on it instead of polling: a sync that polls a boolean is a sync that cannot stop
// promptly, and a source bridging to context.Context needs the channel anyway.
type Context interface {
	// Done is closed when the request is cancelled.
	Done() <-chan struct{}
	// Err explains a cancellation, or is empty.
	Err() string
}

// cancelable is the context a domain hands its source for one request.
type cancelable struct {
	done      chan struct{}
	once      sync.Once
	mu        sync.Mutex
	cancelled bool
	reason    string
}

// newCancelable builds an open context.
func newCancelable() *cancelable {
	return &cancelable{done: make(chan struct{})}
}

// Done implements Context.
func (c *cancelable) Done() <-chan struct{} { return c.done }

// Err implements Context.
func (c *cancelable) Err() string {
	c.mu.Lock()
	defer c.mu.Unlock()

	return c.reason
}

// cancel stops the request, once.
func (c *cancelable) cancel(reason string) {
	c.once.Do(func() {
		c.mu.Lock()
		c.cancelled = true
		c.reason = reason
		c.mu.Unlock()

		close(c.done)
	})
}

// cancelled reports whether the request was cancelled.
func (c *cancelable) cancelledNow() bool {
	select {
	case <-c.done:
		return true
	default:
		return false
	}
}

// Data is what the domain needs at spawn.
type Data struct {
	// Source is the chain to serve. Fixed at spawn like every other input.
	Source Source
}

// Chans is the domain's fixed channel set.
type Chans struct {
	// Link carries frames both ways.
	Link tinyspawn.Duplex[[]byte]
	// Stop cancels the domain and everything in flight.
	Stop tinyspawn.Breaker
}

// Serve is the chain-data domain: an event loop over one duplex.
//
// Cancellation arrives while a request is in flight, so the frames are read by a small
// relay that owns the receive side and hands requests to this loop. The relay cancels a
// request's context directly, which is the one thing that must not wait for the loop: a sync
// that is being cancelled is by definition the one the loop is inside.
func Serve(data Data, chans Chans) {
	requests := make(chan Frame, 4)
	relayDone := make(chan struct{})

	inFlight := &registry{}

	go relay(chans.Link, chans.Stop, requests, inFlight, relayDone)

	for {
		select {
		case <-chans.Stop.Done():
			return

		case <-relayDone:
			return

		case frame, ok := <-requests:
			if !ok {
				return
			}

			answer := handle(data.Source, frame, inFlight)

			select {
			case chans.Link.Tx <- answer:
			case <-chans.Stop.Done():
				return
			case <-relayDone:
				// The peer is gone; there is nobody to answer.
				return
			}
		}
	}
}

// registry tracks the request currently in flight, so a cancel frame can reach it.
type registry struct {
	mu      sync.Mutex
	current *cancelable
	id      uint32
}

// put registers a request.
func (r *registry) put(id uint32, ctx *cancelable) {
	r.mu.Lock()
	defer r.mu.Unlock()

	r.current = ctx
	r.id = id
}

// clear forgets a request.
func (r *registry) clear(id uint32) {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.id == id {
		r.current = nil
		r.id = 0
	}
}

// cancel stops the registered request if its id matches.
func (r *registry) cancel(id uint32, reason string) {
	r.mu.Lock()
	current, held := r.current, r.id
	r.mu.Unlock()

	if current != nil && held == id {
		current.cancel(reason)
	}
}

// relay reads frames and routes them, so cancellation is never blocked behind the work.
func relay(
	link tinyspawn.Duplex[[]byte],
	stop tinyspawn.Breaker,
	requests chan<- Frame,
	inFlight *registry,
	done chan<- struct{},
) {
	defer close(done)
	defer close(requests)

	for {
		select {
		case <-stop.Done():
			return

		case raw, ok := <-link.Recv():
			if !ok {
				// The transaction port closed, which is what a terminated worker looks
				// like. Everything in flight fails.
				inFlight.cancel(0, "the channel closed")

				return
			}

			frame, err := Decode(raw)

			if err != nil {
				// A frame this version cannot read is not a request. Answer it so the
				// sender is not left waiting, but never block the relay on it.
				select {
				case link.Tx <- EncodeOrClose(StatusMalformed, err.Error()):
				case <-stop.Done():
					return
				default:
				}

				continue
			}

			switch frame.Kind {
			case KindCancel:
				inFlight.cancel(frame.ID, "cancelled by the wallet")

			case KindClose:
				inFlight.cancel(0, "the wallet closed the channel")

				return

			case KindRequest:
				requests <- frame
			}
		}
	}
}

// encodeResult builds an answer frame.
func encodeResult(frame Frame, status Status, payload []byte, message string) []byte {
	out := Frame{
		Kind: KindResult, ID: frame.ID, Method: frame.Method, Status: status,
		Payload: payload,
	}

	if status != StatusOK && message != "" {
		out.Payload = append([]byte(message), 0)
	}

	encoded, err := Encode(out)

	if err != nil {
		// The only way this fails is an over-large payload, which the codecs bound before
		// they build one; answer with the failure rather than sending nothing.
		fallback, _ := Encode(Frame{
			Kind: KindResult, ID: frame.ID, Method: frame.Method, Status: StatusMalformed,
			Payload: []byte("chaindata: the answer was too large\x00"),
		})

		return fallback
	}

	return encoded
}

// EncodeOrClose is an answer with no request to answer, for a transport-level failure.
func EncodeOrClose(status Status, message string) []byte {
	encoded, _ := Encode(Frame{
		Kind: KindResult, Status: status, Payload: append([]byte(message), 0),
	})

	return encoded
}

// handle runs one request.
func handle(source Source, frame Frame, inFlight *registry) []byte {
	if source == nil {
		return encodeResult(frame, StatusFailed, nil, "this chain-data domain has no source")
	}

	ctx := newCancelable()

	inFlight.put(frame.ID, ctx)

	defer inFlight.clear(frame.ID)

	switch frame.Method {
	case MethodStatus:
		position, err := source.Position(ctx)

		return positionResult(frame, ctx, position, err)

	case MethodSync:
		if len(frame.Payload) != 4 {
			return encodeResult(frame, StatusMalformed, nil, "sync takes a 4-byte height")
		}

		position, err := source.Sync(ctx, binary.LittleEndian.Uint32(frame.Payload))

		return positionResult(frame, ctx, position, err)

	case MethodMatch:
		scripts, err := ReadScripts(frame.Payload)

		if err != nil {
			return encodeResult(frame, StatusMalformed, nil, err.Error())
		}

		heights, err := source.Match(ctx, scripts)

		if ctx.cancelledNow() {
			return encodeResult(frame, StatusCanceled, nil, ctx.Err())
		}

		if err != nil {
			return encodeResult(frame, StatusFailed, nil, err.Error())
		}

		return encodeResult(frame, StatusOK, PutHeights(nil, heights), "")

	case MethodBlock:
		height, err := ReadHeight(frame.Payload)

		if err != nil {
			return encodeResult(frame, StatusMalformed, nil, err.Error())
		}

		raw, err := source.Block(ctx, height)

		if ctx.cancelledNow() {
			return encodeResult(frame, StatusCanceled, nil, ctx.Err())
		}

		if err != nil {
			return encodeResult(frame, StatusFailed, nil, err.Error())
		}

		return encodeResult(frame, StatusOK, raw, "")

	case MethodBroadcast:
		if len(frame.Payload) == 0 {
			return encodeResult(frame, StatusMalformed, nil, "broadcast takes one raw transaction")
		}

		if err := source.Broadcast(ctx, frame.Payload); err != nil {
			if ctx.cancelledNow() {
				return encodeResult(frame, StatusCanceled, nil, ctx.Err())
			}

			return encodeResult(frame, StatusFailed, nil, err.Error())
		}

		if ctx.cancelledNow() {
			return encodeResult(frame, StatusCanceled, nil, ctx.Err())
		}

		return encodeResult(frame, StatusOK, nil, "")

	case MethodFilters:
		from, count, err := ReadFilterRange(frame.Payload)

		if err != nil {
			return encodeResult(frame, StatusMalformed, nil, err.Error())
		}

		filters, err := source.Filters(ctx, from, count)

		if ctx.cancelledNow() {
			return encodeResult(frame, StatusCanceled, nil, ctx.Err())
		}

		if err != nil {
			return encodeResult(frame, StatusFailed, nil, err.Error())
		}

		payload, err := PutFilters(nil, filters)

		if err != nil {
			return encodeResult(frame, StatusMalformed, nil, err.Error())
		}

		return encodeResult(frame, StatusOK, payload, "")
	}

	return encodeResult(frame, StatusUnknownMethod, nil,
		"unknown chain-data method: "+frame.Method.String())
}

// positionResult encodes a position, or the cancellation that stopped it.
func positionResult(frame Frame, ctx *cancelable, position Position, err error) []byte {
	if ctx.cancelledNow() {
		return encodeResult(frame, StatusCanceled, nil, ctx.Err())
	}

	if err != nil {
		return encodeResult(frame, StatusFailed, nil, err.Error())
	}

	return encodeResult(frame, StatusOK, PutPosition(nil, position), "")
}

//
// The client: the wallet's end of the channel.
//

// Client is the wallet's end of a chain-data channel.
//
// It is one value a wallet's domain holds per chain. Answers arrive on a reader goroutine and
// are routed by request id, so a request can be cancelled and a close can fail everything in
// flight without the caller having to poll.
type Client struct {
	// Name is the chain this client speaks to, for error messages.
	Name string

	link tinyspawn.Duplex[[]byte]
	stop tinyspawn.Breaker

	mu      sync.Mutex
	seq     uint32
	pending map[uint32]chan Frame
	closed  bool
	reason  string
	// timeout is the deadline used when timeoutSet is true; otherwise each method's
	// default applies.
	timeout    time.Duration
	timeoutSet bool
}

// NewClient wraps one end of a duplex and starts its reader.
func NewClient(name string, link tinyspawn.Duplex[[]byte], stop tinyspawn.Breaker) *Client {
	client := &Client{
		Name: name, link: link, stop: stop, pending: map[uint32]chan Frame{},
		timeout: DefaultRequestTimeout,
	}

	go client.reader()

	return client
}

// DefaultRequestTimeout bounds how long a request waits for an answer.
//
// Without it a wallet whose chain channel never came up waits for ever: the page shows
// "opening the workers…" (or a spinner on the first status), nothing is logged anywhere, and a
// dead channel is indistinguishable from a slow one. A timeout turns that into a reason.
//
// It deliberately does not apply to Sync: a first sync is minutes of work in one request, and
// a deadline meant to catch a dead channel must not kill a live one that is making progress.
// Sync is bounded by the channel closing, the caller cancelling it, and the light client's own
// read timeout.
const DefaultRequestTimeout = 60 * time.Second

// ErrNoAnswer means the far end did not answer within the timeout.
var ErrNoAnswer = errors.New("chaindata: the far end did not answer")

// SetTimeout overrides the answer deadline for every method, including Sync. Zero restores the
// per-method defaults, which means Sync has no answer deadline again.
func (c *Client) SetTimeout(timeout time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if timeout <= 0 {
		c.timeout = DefaultRequestTimeout
		c.timeoutSet = false

		return
	}

	c.timeout = timeout
	c.timeoutSet = true
}

// reader routes answers to the requests waiting for them.
func (c *Client) reader() {
	for {
		select {
		case <-c.stop.Done():
			c.failAll("the wallet stopped")

			return

		case raw, ok := <-c.link.Recv():
			if !ok {
				c.failAll("the chain-data channel closed")

				return
			}

			frame, err := Decode(raw)

			if err != nil {
				// An answer this version cannot read cannot be matched to a request. The
				// request it belonged to will time out with the caller's own deadline
				// rather than being answered with somebody else's answer.
				continue
			}

			if frame.Kind == KindClose {
				c.failAll(reasonOf(frame, "the chain-data domain closed"))

				return
			}

			c.deliver(frame)
		}
	}
}

// deliver hands one answer to its waiter, if there is one.
func (c *Client) deliver(frame Frame) {
	c.mu.Lock()

	waiter, ok := c.pending[frame.ID]

	if ok {
		delete(c.pending, frame.ID)
	}

	c.mu.Unlock()

	if !ok {
		// An answer to a request nobody is waiting for: a cancel raced the answer, or the
		// request was abandoned. Dropping it is the only honest option.
		return
	}

	select {
	case waiter <- frame:
	default:
	}
}

// failAll fails everything in flight and refuses new requests.
func (c *Client) failAll(reason string) {
	c.mu.Lock()

	c.closed = true
	c.reason = reason

	waiting := c.pending
	c.pending = map[uint32]chan Frame{}

	c.mu.Unlock()

	for _, waiter := range waiting {
		select {
		case waiter <- Frame{Kind: KindClose, Status: StatusClosed,
			Payload: append([]byte(reason), 0)}:
		default:
		}
	}
}

// reasonOf reads the message a close frame carries.
func reasonOf(frame Frame, fallback string) string {
	if len(frame.Payload) == 0 {
		return fallback
	}

	message := string(frame.Payload)

	return trimNul(message)
}

// trimNul cuts the trailing NUL a message payload carries.
func trimNul(text string) string {
	if n := len(text); n > 0 && text[n-1] == 0 {
		return text[:n-1]
	}

	return text
}

// Closed reports whether the channel is gone.
func (c *Client) Closed() bool {
	c.mu.Lock()
	defer c.mu.Unlock()

	return c.closed
}

// Do sends one request and waits for its answer.
//
// The wait is bounded by the breaker and by the channel closing, not by a deadline: a caller
// with a deadline should cancel the request, which sends a frame the far end acts on.
func (c *Client) Do(frame Frame) (Frame, error) {
	c.mu.Lock()

	if c.closed {
		reason := c.reason
		c.mu.Unlock()

		return Frame{}, fmt.Errorf("%s: %s", c.Name, reason)
	}

	c.seq++
	id := c.seq

	waiter := make(chan Frame, 1)
	c.pending[id] = waiter

	c.mu.Unlock()

	frame.Kind = KindRequest
	frame.ID = id

	encoded, err := Encode(frame)

	if err != nil {
		c.mu.Lock()
		delete(c.pending, id)
		c.mu.Unlock()

		return Frame{}, fmt.Errorf("%s: %w", c.Name, err)
	}

	select {
	case c.link.Tx <- encoded:
	case <-c.stop.Done():
		c.mu.Lock()
		delete(c.pending, id)
		c.mu.Unlock()

		return Frame{}, fmt.Errorf("%s: the wallet stopped", c.Name)
	}

	// The deadline is computed once, for this method: a sync may legitimately take minutes,
	// while a status that takes a minute is a dead channel. A zero deadline leaves timeoutCh
	// nil, which disables that case in the select below.
	timeout := c.requestTimeout(frame.Method)

	var timeoutCh <-chan time.Time

	if timeout > 0 {
		timer := time.NewTimer(timeout)
		defer timer.Stop()

		timeoutCh = timer.C
	}

	select {
	case answer := <-waiter:
		if answer.Kind == KindClose {
			return Frame{}, fmt.Errorf("%s: %s", c.Name, reasonOf(answer, "closed"))
		}

		if answer.Status != StatusOK {
			return answer, fmt.Errorf("%s: %s", c.Name, statusMessage(answer))
		}

		return answer, nil

	case <-timeoutCh:
		c.mu.Lock()
		delete(c.pending, id)
		c.mu.Unlock()

		return Frame{}, fmt.Errorf(
			"%s: %w within %s: is the chain channel open?",
			c.Name, ErrNoAnswer, timeout.Round(time.Second),
		)

	case <-c.stop.Done():
		c.mu.Lock()
		delete(c.pending, id)
		c.mu.Unlock()

		return Frame{}, fmt.Errorf("%s: the wallet stopped", c.Name)
	}
}

// requestTimeout is the answer deadline for one method, or zero for no deadline.
//
// An explicit SetTimeout applies to every method, Sync included. Otherwise only Sync is
// unbounded: it is one long piece of work, and its liveness is the channel's own (a close, the
// caller's cancellation, the light client's read timeout), not an answer deadline. A deadline
// meant to catch a dead channel must not kill a live sync that is making progress.
func (c *Client) requestTimeout(method Method) time.Duration {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.timeoutSet {
		return c.timeout
	}

	if method == MethodSync {
		return 0
	}

	return DefaultRequestTimeout
}

// Cancel asks the far end to abandon one request.
//
// It is best effort by design: the request may have finished already, and a cancel frame for
// an answer that is on its way is dropped by the far end.
func (c *Client) Cancel(id uint32) {
	encoded, err := Encode(Frame{Kind: KindCancel, ID: id})

	if err != nil {
		return
	}

	select {
	case c.link.Tx <- encoded:
	case <-c.stop.Done():
	default:
	}
}

// CancelPending cancels everything in flight. The clear is done under the same lock a
// finished request takes, so a cancel cannot be sent twice for one id.
func (c *Client) CancelPending() {
	c.mu.Lock()
	ids := make([]uint32, 0, len(c.pending))

	for id := range c.pending {
		ids = append(ids, id)
	}

	c.mu.Unlock()

	for _, id := range ids {
		c.Cancel(id)
	}
}

// Close sends a close frame and stops the client.
//
// The frame is what tells the far end to finish; closing the breaker is what tells this end's
// reader to stop. Both are needed: a domain that only stopped reading would leave the other
// side working.
func (c *Client) Close() {
	encoded, err := Encode(Frame{Kind: KindClose, Status: StatusClosed})

	if err == nil {
		select {
		case c.link.Tx <- encoded:
		case <-c.stop.Done():
		default:
		}
	}

	if c.stop != nil && !c.stop.Closed() {
		c.stop.Close()
	}
}

// statusMessage reads the message a failed result carries.
func statusMessage(frame Frame) string {
	if len(frame.Payload) == 0 {
		return frame.Status.String()
	}

	return trimNul(string(frame.Payload))
}

// Position asks the chain where it is.
func (c *Client) Position() (Position, error) {
	answer, err := c.Do(Frame{Method: MethodStatus})

	if err != nil {
		return Position{}, err
	}

	return ReadPosition(answer.Payload)
}

// Sync advances the chain from a height.
func (c *Client) Sync(from uint32) (Position, error) {
	answer, err := c.Do(Frame{
		Method:  MethodSync,
		Payload: binary.LittleEndian.AppendUint32(nil, from),
	})

	if err != nil {
		return Position{}, err
	}

	return ReadPosition(answer.Payload)
}

// Match asks which heights mention any of the scripts.
func (c *Client) Match(scripts [][]byte) ([]uint32, error) {
	payload, err := PutScripts(nil, scripts)

	if err != nil {
		return nil, err
	}

	answer, err := c.Do(Frame{Method: MethodMatch, Payload: payload})

	if err != nil {
		return nil, err
	}

	return ReadHeights(answer.Payload)
}

// Filters reads a contiguous run of compact filters.
func (c *Client) Filters(from, count uint32) ([]Filter, error) {
	answer, err := c.Do(Frame{
		Method:  MethodFilters,
		Payload: PutFilterRange(nil, from, count),
	})

	if err != nil {
		return nil, err
	}

	return ReadFilters(answer.Payload)
}

// Block reads one raw block by height.
//
// The bytes are unverified here on purpose: this is the transport, and the proof is the merkle
// root in nostrino/pkg/block, checked by whoever knows which header the block is supposed to
// belong to.
func (c *Client) Block(height uint32) ([]byte, error) {
	answer, err := c.Do(Frame{Method: MethodBlock, Payload: PutHeight(nil, height)})

	if err != nil {
		return nil, err
	}

	return answer.Payload, nil
}

// Broadcast hands one raw transaction to the chain.
//
// An empty transaction is refused here rather than sent, and a successful answer means the far
// end sent the transaction to its peer; whether the transaction is in a mempool is a different
// fact and has to be checked on chain.
func (c *Client) Broadcast(raw []byte) error {
	if len(raw) == 0 {
		return errors.New("chaindata: refusing to broadcast an empty transaction")
	}

	_, err := c.Do(Frame{Method: MethodBroadcast, Payload: raw})

	return err
}