Protocol/stream.go

package eq2net

import (
	"encoding/binary"
	"fmt"
	"net"
	"sync"
	"time"

	"git.sharkk.net/EQ2/Protocol/crypto"
	"github.com/panjf2000/gnet/v2"
)

// StreamState represents the current state of an EQStream connection
type StreamState int

const (
	CLOSED        StreamState = iota // No connection
	CONNECTING                       // Session request sent, awaiting response
	ESTABLISHED                      // Active connection ready for data
	DISCONNECTING                    // Graceful disconnect in progress
)

// StreamType identifies the type of EQ stream
type StreamType int

const (
	UnknownStream StreamType = iota
	LoginStream
	WorldStream
	ZoneStream
	ChatStream
	VoiceStream
)

// Stream configuration constants
const (
	MaxPacketSize         = 512   // Maximum packet size
	MaxCombinedSize       = 255   // Maximum combined packet size
	RetransmitTimeoutMin  = 500   // Minimum retransmit timeout (ms)
	RetransmitTimeoutMax  = 5000  // Maximum retransmit timeout (ms)
	RetransmitMaxAttempts = 10    // Maximum retransmission attempts
	KeepAliveInterval     = 30000 // Keep-alive interval (ms)
	SessionTimeout        = 90000 // Session timeout (ms)

	// Packet flags
	FLAG_COMPRESSED = 0x01 // Packet is compressed
	FLAG_ENCODED    = 0x04 // Packet is encoded/encrypted

	// Rate limiting
	RATEBASE  = 1048576 // Base rate: 1 MB
	DECAYBASE = 78642   // Decay rate: RATEBASE/10
)

// RetransmitEntry tracks packets awaiting acknowledgment
type RetransmitEntry struct {
	packet   *ProtocolPacket
	sentTime time.Time
	attempts int
}

// EQStream manages a single EverQuest network stream connection
type EQStream struct {
	// Network connection
	conn       gnet.Conn
	remoteIP   net.IP
	remotePort uint16

	// Session identification
	sessionID  uint32
	streamType StreamType

	// State management
	state   StreamState
	stateMu sync.RWMutex

	// Encryption
	crypto    *crypto.Ciphers
	cryptoKey uint32

	// Compression settings
	compressed bool
	encoded    bool

	// Sequence numbers
	nextInSeq  uint16
	nextOutSeq uint16
	lastAckSeq uint16
	seqMu      sync.Mutex

	// Packet queues
	outbound      chan *ProtocolPacket
	retransmit    map[uint16]*RetransmitEntry
	retransmitMu  sync.RWMutex
	futurePackets map[uint16]*ProtocolPacket
	futureMu      sync.Mutex

	// Combined packet handling
	combinedApp  *EQPacket
	combinedMu   sync.Mutex
	combineTimer *time.Timer

	// Flow control
	currentRate   uint32
	rateThreshold uint32
	rateMu        sync.Mutex

	// Timing
	lastActivity time.Time
	lastPing     time.Time
	avgDelta     time.Duration
	timingMu     sync.RWMutex

	// Statistics
	packetsReceived uint64
	packetsSent     uint64
	bytesReceived   uint64
	bytesSent       uint64
	statsMu         sync.RWMutex

	// Configuration
	opcodeSize    uint8
	maxPacketSize uint32
	Version       int16 // Client version for opcode conversion

	// Callbacks
	onEmuPacket  func(*EQPacket) // Called when emulator packet is ready
	onDisconnect func()          // Called on disconnect
}

// NewEQStream creates a new EQ stream
func NewEQStream(conn gnet.Conn) *EQStream {
	stream := &EQStream{
		conn:          conn,
		state:         CLOSED,
		sessionID:     0,
		streamType:    UnknownStream,
		compressed:    true,
		encoded:       false,
		nextInSeq:     0,
		nextOutSeq:    0,
		lastAckSeq:    0,
		outbound:      make(chan *ProtocolPacket, 1024),
		retransmit:    make(map[uint16]*RetransmitEntry),
		futurePackets: make(map[uint16]*ProtocolPacket),
		currentRate:   RATEBASE,
		rateThreshold: DECAYBASE,
		lastActivity:  time.Now(),
		opcodeSize:    2,
		maxPacketSize: MaxPacketSize,
		cryptoKey:     0x33624702, // Default CRC key
	}

	// Parse remote address
	if addr := conn.RemoteAddr(); addr != nil {
		if tcpAddr, ok := addr.(*net.TCPAddr); ok {
			stream.remoteIP = tcpAddr.IP
			stream.remotePort = uint16(tcpAddr.Port)
		} else if udpAddr, ok := addr.(*net.UDPAddr); ok {
			stream.remoteIP = udpAddr.IP
			stream.remotePort = uint16(udpAddr.Port)
		}
	}

	return stream
}

// GetState returns the current stream state
func (s *EQStream) GetState() StreamState {
	s.stateMu.RLock()
	defer s.stateMu.RUnlock()
	return s.state
}

// SetState updates the stream state
func (s *EQStream) SetState(state StreamState) {
	s.stateMu.Lock()
	defer s.stateMu.Unlock()
	s.state = state
}

// IsActive returns true if the stream is established
func (s *EQStream) IsActive() bool {
	return s.GetState() == ESTABLISHED
}

// IsClosed returns true if the stream is closed
func (s *EQStream) IsClosed() bool {
	return s.GetState() == CLOSED
}

// ProcessPacket processes an incoming protocol packet
func (s *EQStream) ProcessPacket(p *ProtocolPacket) error {
	s.updateActivity()
	s.updateStats(true, uint64(p.TotalSize()))

	// Handle based on opcode
	switch p.Opcode {
	case OP_SessionRequest:
		return s.handleSessionRequest(p)
	case OP_SessionResponse:
		return s.handleSessionResponse(p)
	case OP_SessionDisconnect:
		return s.handleDisconnect(p)
	case OP_KeepAlive:
		return s.handleKeepAlive(p)
	case OP_Ack:
		return s.handleAck(p)
	case OP_OutOfOrderAck:
		return s.handleOutOfOrderAck(p)
	case OP_Packet:
		return s.handleDataPacket(p)
	case OP_Fragment:
		return s.handleFragment(p)
	case OP_Combined:
		return s.handleCombined(p)
	case OP_AppCombined:
		return s.handleAppCombined(p)
	default:
		return fmt.Errorf("unknown opcode: 0x%04X", p.Opcode)
	}
}

// handleSessionRequest processes a session request packet
func (s *EQStream) handleSessionRequest(p *ProtocolPacket) error {
	if len(p.Buffer) < 4 {
		return fmt.Errorf("session request too small")
	}

	// Parse session ID
	s.sessionID = binary.BigEndian.Uint32(p.Buffer[0:4])

	// Send session response
	s.sendSessionResponse()

	// Update state
	s.SetState(ESTABLISHED)

	return nil
}

// handleSessionResponse processes a session response packet
func (s *EQStream) handleSessionResponse(p *ProtocolPacket) error {
	if len(p.Buffer) < 4 {
		return fmt.Errorf("session response too small")
	}

	// Parse session parameters
	s.sessionID = binary.BigEndian.Uint32(p.Buffer[0:4])

	// Parse compression/encoding flags if present
	if len(p.Buffer) >= 5 {
		flags := p.Buffer[4]
		s.compressed = (flags & FLAG_COMPRESSED) != 0
		s.encoded = (flags & FLAG_ENCODED) != 0
	}

	// Initialize encryption if needed
	if s.encoded && s.crypto == nil {
		// This would be initialized with proper key exchange
		// For now, using default key
		var err error
		s.crypto, err = crypto.NewCiphers(int64(s.cryptoKey))
		if err != nil {
			return fmt.Errorf("failed to initialize encryption: %w", err)
		}
	}

	// Update state
	s.SetState(ESTABLISHED)

	// Send keep alive to confirm
	s.sendKeepAlive()

	return nil
}

// handleDisconnect processes a disconnect packet
func (s *EQStream) handleDisconnect(p *ProtocolPacket) error {
	s.SetState(CLOSED)

	// Call disconnect callback if set
	if s.onDisconnect != nil {
		s.onDisconnect()
	}

	return nil
}

// handleKeepAlive processes a keep-alive packet
func (s *EQStream) handleKeepAlive(p *ProtocolPacket) error {
	// Update activity timestamp
	s.updateActivity()

	// Send keep-alive response if needed
	// Some implementations echo the keep-alive

	return nil
}

// handleAck processes an acknowledgment packet
func (s *EQStream) handleAck(p *ProtocolPacket) error {
	if len(p.Buffer) < 2 {
		return fmt.Errorf("ack packet too small")
	}

	// Parse acknowledged sequence number
	ackSeq := binary.BigEndian.Uint16(p.Buffer[0:2])

	// Remove from retransmit queue
	s.retransmitMu.Lock()
	delete(s.retransmit, ackSeq)
	s.retransmitMu.Unlock()

	// Update last acknowledged sequence
	s.seqMu.Lock()
	if s.sequenceGreaterThan(ackSeq, s.lastAckSeq) {
		s.lastAckSeq = ackSeq
	}
	s.seqMu.Unlock()

	return nil
}

// handleOutOfOrderAck processes an out-of-order acknowledgment
func (s *EQStream) handleOutOfOrderAck(p *ProtocolPacket) error {
	// Similar to handleAck but indicates out-of-order reception
	return s.handleAck(p)
}

// handleDataPacket processes a data packet
func (s *EQStream) handleDataPacket(p *ProtocolPacket) error {
	// Check sequence number
	seq := p.Sequence

	s.seqMu.Lock()
	expectedSeq := s.nextInSeq
	s.seqMu.Unlock()

	seqOrder := s.compareSequence(expectedSeq, seq)

	switch seqOrder {
	case SeqInOrder:
		// Process packet immediately
		s.seqMu.Lock()
		s.nextInSeq++
		s.seqMu.Unlock()

		// Send ACK
		s.sendAck(seq)

		// Extract emulator packet and queue
		if emuPacket := p.ExtractEmuPacket(s.opcodeSize); emuPacket != nil {
			// Process based on stream type
			s.processIncomingEmuPacket(emuPacket)
		}

		// Check for queued future packets
		s.processFuturePackets()

	case SeqFuture:
		// Queue for later processing
		s.futureMu.Lock()
		s.futurePackets[seq] = p
		s.futureMu.Unlock()

		// Send out-of-order ACK
		s.sendOutOfOrderAck(seq)

	case SeqPast:
		// Duplicate packet, just ACK it
		s.sendAck(seq)
	}

	return nil
}

// handleFragment processes a fragmented packet
func (s *EQStream) handleFragment(p *ProtocolPacket) error {
	// TODO: Implement fragment reassembly
	// This requires tracking fragment sequences and reassembling
	return fmt.Errorf("fragment handling not yet implemented")
}

// handleCombined processes a combined packet
func (s *EQStream) handleCombined(p *ProtocolPacket) error {
	// Extract sub-packets
	subPackets, err := p.ExtractSubPackets()
	if err != nil {
		return fmt.Errorf("failed to extract sub-packets: %w", err)
	}

	// Process each sub-packet
	for _, subPacket := range subPackets {
		if err := s.ProcessPacket(subPacket); err != nil {
			// Log error but continue processing other packets
			continue
		}
	}

	return nil
}

// handleAppCombined processes an application-level combined packet
func (s *EQStream) handleAppCombined(p *ProtocolPacket) error {
	// For now, just try to extract as single packet
	// TODO: Implement proper app-level packet extraction
	emuPacket := p.ExtractEmuPacket(s.opcodeSize)
	if emuPacket == nil {
		return fmt.Errorf("failed to extract emu packet")
	}

	// Process the packet
	s.processIncomingEmuPacket(emuPacket)

	return nil
}

// QueuePacket queues an outbound packet for transmission
func (s *EQStream) QueuePacket(p *ProtocolPacket, reliable bool) error {
	if s.GetState() != ESTABLISHED {
		return fmt.Errorf("stream not established")
	}

	// Add sequence number for reliable packets
	if reliable {
		s.seqMu.Lock()
		p.Sequence = s.nextOutSeq
		s.nextOutSeq++
		s.seqMu.Unlock()

		// Add to retransmit queue
		s.retransmitMu.Lock()
		s.retransmit[p.Sequence] = &RetransmitEntry{
			packet:   p,
			sentTime: time.Now(),
			attempts: 0,
		}
		s.retransmitMu.Unlock()
	}

	// Add CRC if needed
	if p.Opcode != OP_SessionRequest && p.Opcode != OP_SessionResponse {
		s.addCRC(p)
	}

	// Compress if beneficial
	if s.compressed && len(p.Buffer) > 30 {
		p.Compress()
	}

	// Encrypt if enabled
	if s.encoded && s.crypto != nil {
		s.encryptPacket(p)
	}

	// Send packet
	return s.sendPacket(p)
}

// Helper methods

func (s *EQStream) updateActivity() {
	s.timingMu.Lock()
	s.lastActivity = time.Now()
	s.timingMu.Unlock()
}

func (s *EQStream) updateStats(received bool, bytes uint64) {
	s.statsMu.Lock()
	if received {
		s.packetsReceived++
		s.bytesReceived += bytes
	} else {
		s.packetsSent++
		s.bytesSent += bytes
	}
	s.statsMu.Unlock()
}

func (s *EQStream) sendPacket(p *ProtocolPacket) error {
	data := p.Serialize()
	s.updateStats(false, uint64(len(data)))
	return s.conn.AsyncWrite(data, nil)
}

func (s *EQStream) sendSessionResponse() {
	resp := NewProtocolPacket(OP_SessionResponse, nil)

	// Build response data
	data := make([]byte, 5)
	binary.BigEndian.PutUint32(data[0:4], s.sessionID)

	// Set flags
	flags := byte(0)
	if s.compressed {
		flags |= FLAG_COMPRESSED
	}
	if s.encoded {
		flags |= FLAG_ENCODED
	}
	data[4] = flags

	resp.Buffer = data
	s.sendPacket(resp)
}

func (s *EQStream) sendKeepAlive() {
	ka := NewProtocolPacket(OP_KeepAlive, nil)
	s.sendPacket(ka)

	s.timingMu.Lock()
	s.lastPing = time.Now()
	s.timingMu.Unlock()
}

func (s *EQStream) sendAck(seq uint16) {
	ack := NewProtocolPacket(OP_Ack, nil)
	ack.Buffer = make([]byte, 2)
	binary.BigEndian.PutUint16(ack.Buffer, seq)
	s.sendPacket(ack)
}

func (s *EQStream) sendOutOfOrderAck(seq uint16) {
	ack := NewProtocolPacket(OP_OutOfOrderAck, nil)
	ack.Buffer = make([]byte, 2)
	binary.BigEndian.PutUint16(ack.Buffer, seq)
	s.sendPacket(ack)
}

func (s *EQStream) addCRC(p *ProtocolPacket) {
	// Serialize packet without CRC
	data := p.Serialize()

	// Calculate CRC
	crc := crypto.CalculateCRC(data, s.cryptoKey)

	// Append CRC to buffer
	crcBytes := make([]byte, 2)
	binary.BigEndian.PutUint16(crcBytes, crc)
	p.Buffer = append(p.Buffer, crcBytes...)
}

func (s *EQStream) encryptPacket(p *ProtocolPacket) {
	if s.crypto == nil {
		return
	}

	// Encrypt the buffer (skip opcode)
	if len(p.Buffer) > 2 {
		s.crypto.Encrypt(p.Buffer[2:])
		p.Encrypted = true
	}
}

func (s *EQStream) processFuturePackets() {
	s.futureMu.Lock()
	defer s.futureMu.Unlock()

	s.seqMu.Lock()
	expectedSeq := s.nextInSeq
	s.seqMu.Unlock()

	// Check if we have the next expected packet
	if p, ok := s.futurePackets[expectedSeq]; ok {
		delete(s.futurePackets, expectedSeq)

		// Process it (this will recursively check for more)
		go s.ProcessPacket(p)
	}
}

// Sequence number comparison

type SeqOrder int

const (
	SeqPast    SeqOrder = iota // Sequence is from the past
	SeqInOrder                 // Sequence is expected
	SeqFuture                  // Sequence is from the future
)

func (s *EQStream) compareSequence(expected, received uint16) SeqOrder {
	if received == expected {
		return SeqInOrder
	}

	// Handle wraparound
	if s.sequenceGreaterThan(received, expected) {
		return SeqFuture
	}

	return SeqPast
}

func (s *EQStream) sequenceGreaterThan(a, b uint16) bool {
	// Handle sequence number wraparound
	// If the difference is more than half the sequence space,
	// assume wraparound occurred
	diff := int32(a) - int32(b)
	if diff < 0 {
		diff = -diff
	}

	if diff > 32768 { // Half of uint16 max
		return a < b
	}

	return a > b
}

// CheckRetransmissions checks for packets that need retransmission
func (s *EQStream) CheckRetransmissions() {
	now := time.Now()

	s.retransmitMu.Lock()
	defer s.retransmitMu.Unlock()

	for seq, entry := range s.retransmit {
		// Calculate timeout based on average delta and attempts
		timeout := time.Duration(RetransmitTimeoutMin) * time.Millisecond
		if s.avgDelta > 0 {
			timeout = s.avgDelta * 3
		}

		// Exponential backoff
		for i := 0; i < entry.attempts; i++ {
			timeout *= 2
			if timeout > time.Duration(RetransmitTimeoutMax)*time.Millisecond {
				timeout = time.Duration(RetransmitTimeoutMax) * time.Millisecond
				break
			}
		}

		if now.Sub(entry.sentTime) > timeout {
			if entry.attempts >= RetransmitMaxAttempts {
				// Give up on this packet
				delete(s.retransmit, seq)
				continue
			}

			// Retransmit
			entry.attempts++
			entry.sentTime = now
			s.sendPacket(entry.packet)
		}
	}
}

// CheckTimeout checks if the stream has timed out
func (s *EQStream) CheckTimeout() bool {
	s.timingMu.RLock()
	lastActivity := s.lastActivity
	s.timingMu.RUnlock()

	if time.Since(lastActivity) > time.Duration(SessionTimeout)*time.Millisecond {
		s.SetState(CLOSED)
		if s.onDisconnect != nil {
			s.onDisconnect()
		}
		return true
	}

	// Send keep-alive if needed
	if time.Since(lastActivity) > time.Duration(KeepAliveInterval)*time.Millisecond {
		s.sendKeepAlive()
	}

	return false
}

// SendDisconnect sends a disconnect packet
func (s *EQStream) SendDisconnect() {
	disc := NewProtocolPacket(OP_SessionDisconnect, nil)
	s.sendPacket(disc)
	s.SetState(DISCONNECTING)
}

// Close closes the stream
func (s *EQStream) Close() {
	s.SendDisconnect()
	s.SetState(CLOSED)

	// Clean up resources
	close(s.outbound)

	if s.combineTimer != nil {
		s.combineTimer.Stop()
	}
}

// EQ2-specific packet preparation methods

// PreparePacketForWire prepares an application packet for transmission
// This adds EQ2-specific headers like sequence numbers and compression flags
func (s *EQStream) PreparePacketForWire(app *EQPacket, maxLen int16) (*ProtocolPacket, error) {
	// Convert emulator opcode to network opcode based on version
	// This would use an opcode manager in a full implementation
	networkOpcode := app.EmuOpcode

	// Build the wire format
	var buf []byte

	// Add sequence field placeholder (2 bytes)
	buf = append(buf, 0x00, 0x00)

	// Add compression flag placeholder (1 byte)
	buf = append(buf, 0x00)

	// Add opcode with special encoding
	if networkOpcode >= 255 {
		// Oversized opcode
		buf = append(buf, 0xFF) // Marker
		buf = append(buf, byte(networkOpcode>>8), byte(networkOpcode))
	} else {
		buf = append(buf, byte(networkOpcode))
	}

	// Add packet data
	buf = append(buf, app.Buffer...)

	// Create protocol packet
	proto := NewProtocolPacket(OP_Packet, buf)
	proto.Version = app.Version

	return proto, nil
}

// CombinePacketsForWire combines multiple packets at the application level
// This is different from protocol-level combining
func (s *EQStream) CombinePacketsForWire(packets []*EQPacket) *ProtocolPacket {
	if len(packets) == 0 {
		return nil
	}

	// If only one packet, just prepare it normally
	if len(packets) == 1 {
		proto, _ := s.PreparePacketForWire(packets[0], MaxPacketSize)
		return proto
	}

	// Build combined buffer with EQ2 format
	var buf []byte

	for _, app := range packets {
		// Prepare each packet
		prepared, err := s.PreparePacketForWire(app, MaxPacketSize)
		if err != nil {
			continue
		}

		data := prepared.Buffer
		size := len(data)

		// Add size encoding
		if size >= 255 {
			// Oversized packet
			buf = append(buf, 0xFF)
			buf = append(buf, byte(size>>8), byte(size))
		} else {
			buf = append(buf, byte(size))
		}

		// Add packet data
		buf = append(buf, data...)
	}

	// Create combined protocol packet
	return NewProtocolPacket(OP_AppCombined, buf)
}

// QueueAppPacket queues an application packet for transmission
// This handles the full preparation pipeline
func (s *EQStream) QueueAppPacket(app *EQPacket) error {
	// Convert to protocol packet with EQ2 preparation
	proto, err := s.PreparePacketForWire(app, MaxPacketSize)
	if err != nil {
		return err
	}

	// Queue for transmission with reliability
	return s.QueuePacket(proto, true)
}

// QueueLoginPacket queues a login packet for transmission
func (s *EQStream) QueueLoginPacket(p *LoginPacket) error {
	if s.streamType != LoginStream {
		return fmt.Errorf("cannot send login packet on non-login stream")
	}
	
	// Convert to protocol packet
	proto := p.ConvertToProtocolPacket()
	
	// Queue for transmission with reliability
	return s.QueuePacket(proto, true)
}

// QueueGamePacket queues a game packet for transmission
func (s *EQStream) QueueGamePacket(p *GamePacket) error {
	if s.streamType != WorldStream && s.streamType != ZoneStream {
		return fmt.Errorf("cannot send game packet on non-game stream")
	}
	
	// Convert to protocol packet
	proto := p.ConvertToProtocolPacket()
	
	// Queue for transmission with reliability
	return s.QueuePacket(proto, true)
}

// SetOpcodeManager sets the opcode conversion manager for version-specific opcodes
// In a full implementation, this would handle emulator->network opcode conversion
func (s *EQStream) SetOpcodeManager(version int16) {
	// This would set up opcode conversion tables based on client version
	// For now, it's a placeholder
	s.Version = version
}

// processIncomingEmuPacket processes an incoming emulator packet based on stream type
func (s *EQStream) processIncomingEmuPacket(emu *EQPacket) {
	// Parse based on stream type
	switch s.streamType {
	case LoginStream:
		// Parse as login packet
		if _, err := ParseLoginPacket(emu.Buffer); err == nil {
			// For login packets, we might want to do special handling
			// For now, just pass through the emu packet
			if s.onEmuPacket != nil {
				s.onEmuPacket(emu)
			}
		}
		
	case WorldStream, ZoneStream:
		// Parse as game packet  
		if _, err := ParseGamePacket(emu.Buffer); err == nil {
			// For game packets, we might want to do special handling
			// For now, just pass through the emu packet
			if s.onEmuPacket != nil {
				s.onEmuPacket(emu)
			}
		}
		
	default:
		// Unknown stream type, call callback if set
		if s.onEmuPacket != nil {
			s.onEmuPacket(emu)
		}
	}
}