codec_part2.go

package http3

import (
	"errors"

	"gnalloy.org/gnalloy/buffer"
	"gnalloy.org/gnalloy/channel"
	"gnalloy.org/gnalloy/codec"
	"gnalloy.org/gnalloy/message"
)

func readable(buf buffer.ByteBuf) int {
	if buf == nil {
		return 0
	}
	return buf.ReadableBytes()
}

func releaseMessage(msg any) {
	message.Release(msg)
}

func writeBytes(ctx *channel.HandlerContext, data []byte) error {
	out, err := ctx.Channel().Allocator().Acquire(len(data))
	if err != nil {
		return err
	}
	if _, err := out.WriteBytes(data); err != nil {
		out.Release()
		return err
	}
	return codec.WriteOutboundBuffer(ctx, out)
}

func readVarInt(in *buffer.CompositeByteBuf, index int) (uint64, int, bool, error) {
	if in.WriterIndex()-index < 1 {
		return 0, 0, false, nil
	}
	first, ok := in.GetByte(index)
	if !ok {
		return 0, 0, false, nil
	}
	n := 1 << (first >> 6)
	if in.WriterIndex()-index < int(n) {
		return 0, 0, false, nil
	}
	switch n {
	case 1:
		return uint64(first & 0x3f), 1, true, nil
	case 2:
		second, _ := in.GetByte(index + 1)
		return uint64(first&0x3f)<<8 | uint64(second), 2, true, nil
	case 4:
		value := uint64(first&0x3f) << 24
		for i := 1; i < 4; i++ {
			b, _ := in.GetByte(index + i)
			value |= uint64(b) << uint(8*(3-i))
		}
		return value, 4, true, nil
	case 8:
		value := uint64(first&0x3f) << 56
		for i := 1; i < 8; i++ {
			b, _ := in.GetByte(index + i)
			value |= uint64(b) << uint(8*(7-i))
		}
		return value, 8, true, nil
	default:
		return 0, 0, false, ErrInvalidVarInt
	}
}

func appendVarInt(dst []byte, v uint64) ([]byte, error) {
	switch {
	case v <= 63:
		return append(dst, byte(v)), nil
	case v <= 16383:
		return append(dst, byte(v>>8)|0x40, byte(v)), nil
	case v <= 1073741823:
		return append(dst, byte(v>>24)|0x80, byte(v>>16), byte(v>>8), byte(v)), nil
	case v <= maxVarInt:
		return append(dst, byte(v>>56)|0xc0, byte(v>>48), byte(v>>40), byte(v>>32), byte(v>>24), byte(v>>16), byte(v>>8), byte(v)), nil
	default:
		return nil, ErrInvalidVarInt
	}
}

// QPACKEncoderStreamDecoder 解码对端 encoder stream 指令。
type QPACKEncoderStreamDecoder struct {
	*codec.ByteToMessageDecoder
}

// NewQPACKEncoderStreamDecoder 创建 encoder stream 指令解码器。
func NewQPACKEncoderStreamDecoder() *QPACKEncoderStreamDecoder {
	d := &QPACKEncoderStreamDecoder{}
	d.ByteToMessageDecoder = codec.NewByteToMessageDecoder(d)
	return d
}

// Decode 解码一条完整 encoder stream 指令。
func (d *QPACKEncoderStreamDecoder) Decode(_ *channel.HandlerContext, in *buffer.CompositeByteBuf) (any, error) {
	msg, n, ok, err := decodeQPACKEncoderInstruction(in, in.ReaderIndex())
	if err != nil || !ok {
		return nil, err
	}
	if err := in.SkipBytes(n); err != nil {
		return nil, err
	}
	return msg, nil
}

// QPACKEncoderStreamEncoder 编码本端 encoder stream 指令。
type QPACKEncoderStreamEncoder struct{}

// NewQPACKEncoderStreamEncoder 创建 encoder stream 指令编码器。
func NewQPACKEncoderStreamEncoder() *QPACKEncoderStreamEncoder {
	return &QPACKEncoderStreamEncoder{}
}

// Write 编码 encoder stream 指令，其它消息透传。
func (e *QPACKEncoderStreamEncoder) Write(ctx *channel.HandlerContext, msg any) error {
	data, ok, err := appendQPACKEncoderInstruction(nil, msg)
	if err != nil {
		return err
	}
	if !ok {
		return ctx.Write(msg)
	}
	return writeBytes(ctx, data)
}

// QPACKDecoderStreamDecoder 解码对端 decoder stream 反馈指令。
type QPACKDecoderStreamDecoder struct {
	*codec.ByteToMessageDecoder
}

// NewQPACKDecoderStreamDecoder 创建 decoder stream 指令解码器。
func NewQPACKDecoderStreamDecoder() *QPACKDecoderStreamDecoder {
	d := &QPACKDecoderStreamDecoder{}
	d.ByteToMessageDecoder = codec.NewByteToMessageDecoder(d)
	return d
}

// Decode 解码一条完整 decoder stream 指令。
func (d *QPACKDecoderStreamDecoder) Decode(_ *channel.HandlerContext, in *buffer.CompositeByteBuf) (any, error) {
	msg, n, ok, err := decodeQPACKDecoderInstruction(in, in.ReaderIndex())
	if err != nil || !ok {
		return nil, err
	}
	if err := in.SkipBytes(n); err != nil {
		return nil, err
	}
	return msg, nil
}

// QPACKDecoderStreamEncoder 编码本端 decoder stream 反馈指令。
type QPACKDecoderStreamEncoder struct{}

// NewQPACKDecoderStreamEncoder 创建 decoder stream 指令编码器。
func NewQPACKDecoderStreamEncoder() *QPACKDecoderStreamEncoder {
	return &QPACKDecoderStreamEncoder{}
}

// Write 编码 decoder stream 指令，其它消息透传。
func (e *QPACKDecoderStreamEncoder) Write(ctx *channel.HandlerContext, msg any) error {
	data, ok, err := appendQPACKDecoderInstruction(nil, msg)
	if err != nil {
		return err
	}
	if !ok {
		return ctx.Write(msg)
	}
	return writeBytes(ctx, data)
}

// StreamType 是 HTTP/3 单向 QUIC stream 的类型前缀。
type StreamType uint64

const (
	StreamTypeControl      StreamType = 0x00
	StreamTypePush         StreamType = 0x01
	StreamTypeQPACKEncoder StreamType = 0x02
	StreamTypeQPACKDecoder StreamType = 0x03
	// StreamTypeWebTransport 是 WebTransport over HTTP/3 的单向 stream type。
	StreamTypeWebTransport StreamType = 0x54
)

// StreamTypeFrame 表示 QUIC 单向 stream 的首个类型前缀。
type StreamTypeFrame struct {
	Type StreamType
}

// StreamTypeDecoder 从 QUIC 单向 stream 头部解析 HTTP/3 stream type。
type StreamTypeDecoder struct {
	*codec.ByteToMessageDecoder
	seen bool
}

// StreamTypeEncoder 在 QUIC 单向 stream 出站首帧前写入 HTTP/3 stream type。
type StreamTypeEncoder struct {
	streamType StreamType
	wrote      bool
}

// NewStreamTypeDecoder 创建 HTTP/3 stream type decoder。
func NewStreamTypeDecoder() *StreamTypeDecoder {
	d := &StreamTypeDecoder{}
	d.ByteToMessageDecoder = codec.NewByteToMessageListDecoder(d)
	return d
}

// NewStreamTypeEncoder 创建 HTTP/3 stream type encoder。
func NewStreamTypeEncoder(streamType StreamType) *StreamTypeEncoder {
	return &StreamTypeEncoder{streamType: streamType}
}

// DecodeBytes 解码 stream type，并把剩余 frame 字节切片透传给后续 frame decoder。
func (d *StreamTypeDecoder) DecodeBytes(_ *channel.HandlerContext, in *buffer.CompositeByteBuf, out *codec.MessageList) error {
	if !d.seen {
		streamType, n, ok, err := readVarInt(in, in.ReaderIndex())
		if err != nil || !ok {
			return err
		}
		if err := in.SkipBytes(n); err != nil {
			return err
		}
		d.seen = true
		out.Add(StreamTypeFrame{Type: StreamType(streamType)})
	}
	if in.ReadableBytes() == 0 {
		return nil
	}
	payload, err := in.Slice(in.ReaderIndex(), in.ReadableBytes())
	if err != nil {
		return err
	}
	if err := in.SkipBytes(in.ReadableBytes()); err != nil {
		payload.Release()
		return err
	}
	out.Add(payload)
	return nil
}

// Write 确保 stream type 只写一次，后续消息保持原样透传。
func (e *StreamTypeEncoder) Write(ctx *channel.HandlerContext, msg any) error {
	if !e.wrote {
		var prefix []byte
		var err error
		prefix, err = appendVarInt(prefix, uint64(e.streamType))
		if err != nil {
			releaseMessage(msg)
			return err
		}
		if err := writeBytes(ctx, prefix); err != nil {
			releaseMessage(msg)
			return err
		}
		e.wrote = true
	}
	return ctx.Write(msg)
}

// LifecycleEventType 描述 HTTP/3 连接级语义帧事件类型。
type LifecycleEventType uint8

const (
	LifecycleEventSettings LifecycleEventType = iota + 1
	LifecycleEventGoAway
	LifecycleEventMaxPushID
	LifecycleEventCancelPush
	LifecycleEventPriorityUpdate
)

// LifecycleEvent 是 HTTP/3 control/request stream 上可观测的连接级事件。
type LifecycleEvent struct {
	Type               LifecycleEventType
	Settings           []Setting
	ID                 uint64
	PushID             uint64
	PriorityFrameType  FrameType
	PriorityElementID  uint64
	PriorityFieldValue []byte
}

// LifecycleHandler 把 HTTP/3 连接级语义帧转换为 user event，并保留原帧继续传播。
type LifecycleHandler struct{}

// NewLifecycleHandler 创建 HTTP/3 生命周期事件 handler。
func NewLifecycleHandler() *LifecycleHandler {
	return &LifecycleHandler{}
}

// ChannelRead 发布轻量事件；payload 所有权仍由原始 frame 持有。
func (h *LifecycleHandler) ChannelRead(ctx *channel.HandlerContext, msg any) {
	if event, ok := lifecycleEvent(msg); ok {
		ctx.FireUserEventTriggered(event)
	}
	ctx.FireChannelRead(msg)
}

func lifecycleEvent(msg any) (LifecycleEvent, bool) {
	switch frame := msg.(type) {
	case SettingsFrame:
		return LifecycleEvent{Type: LifecycleEventSettings, Settings: cloneSettings(frame.Settings)}, true
	case GoAwayFrame:
		return LifecycleEvent{Type: LifecycleEventGoAway, ID: frame.ID}, true
	case MaxPushIDFrame:
		return LifecycleEvent{Type: LifecycleEventMaxPushID, PushID: frame.PushID}, true
	case CancelPushFrame:
		return LifecycleEvent{Type: LifecycleEventCancelPush, PushID: frame.PushID}, true
	case PriorityUpdateFrame:
		return LifecycleEvent{
			Type:               LifecycleEventPriorityUpdate,
			PriorityFrameType:  frame.Type,
			PriorityElementID:  frame.ElementID,
			PriorityFieldValue: cloneByteBuf(frame.FieldValue),
		}, true
	default:
		return LifecycleEvent{}, false
	}
}

func cloneByteBuf(src buffer.ByteBuf) []byte {
	if src == nil || src.ReadableBytes() == 0 {
		return nil
	}
	out := make([]byte, src.ReadableBytes())
	if buffer.CopyReadableBytes(out, src) != len(out) {
		return nil
	}
	return out
}

var (
	ErrInvalidFrame                = errors.New("http3: invalid frame")
	ErrFrameTooLarge               = errors.New("http3: frame too large")
	ErrInvalidVarInt               = errors.New("http3: invalid varint")
	ErrDuplicateSetting            = errors.New("http3: duplicate setting")
	ErrTooManySettings             = errors.New("http3: too many settings")
	ErrUnsupportedFrame            = errors.New("http3: unsupported frame")
	ErrInvalidFrameOrder           = errors.New("http3: invalid frame order")
	ErrHeaderListTooLarge          = errors.New("http3: header list too large")
	ErrInvalidPipeline             = errors.New("http3: invalid pipeline config")
	ErrQPACKInvalidInstruction     = errors.New("http3: invalid qpack instruction")
	ErrQPACKCapacityExceeded       = errors.New("http3: qpack dynamic table capacity exceeded")
	ErrQPACKEntryTooLarge          = errors.New("http3: qpack dynamic table entry too large")
	ErrQPACKInvalidIndex           = errors.New("http3: invalid qpack dynamic table index")
	ErrQPACKBlockedStreamsExceeded = errors.New("http3: qpack blocked streams exceeded")
	// ErrMissingWebTransportSetting 表示 peer SETTINGS 缺少 WebTransport 必需项。
	ErrMissingWebTransportSetting = errors.New("http3: missing webtransport setting")
	// ErrInvalidWebTransportSetting 表示 peer SETTINGS 中的 WebTransport 开关值非法。
	ErrInvalidWebTransportSetting = errors.New("http3: invalid webtransport setting")
	// ErrInvalidWebTransportConnect 表示 extended CONNECT 头部不是合法 WebTransport 建连请求。
	ErrInvalidWebTransportConnect = errors.New("http3: invalid webtransport connect request")
)