GaCy-GO/internal/gameserver/gateway.go

package gameserver

import (
	"net"
	"sync"
	"sync/atomic"
	"time"

	"gac-hkrpg-go/internal/common"
	kcplib "gac-hkrpg-go/pkg/kcp"
)

// Gateway is the Go port of net/gateway.rs: a raw UDP listener speaking the
// mihoyo KCP handshake (20-byte NetOperation) and dispatching KCP payloads
// to sessions.
type Gateway struct {
	conn     *net.UDPConn
	idSeq    atomic.Uint32
	mu       sync.Mutex
	sessions map[uint32]*Session
}

func NewGateway(host string, port uint16) (*Gateway, error) {
	addr, err := net.ResolveUDPAddr("udp", net.JoinHostPort(host, itoa(int(port))))
	if err != nil {
		return nil, err
	}
	conn, err := net.ListenUDP("udp", addr)
	if err != nil {
		return nil, err
	}
	return &Gateway{conn: conn, sessions: map[uint32]*Session{}}, nil
}

func itoa(v int) string {
	if v == 0 {
		return "0"
	}
	var buf [20]byte
	i := len(buf)
	for v > 0 {
		i--
		buf[i] = byte('0' + v%10)
		v /= 10
	}
	return string(buf[i:])
}

// Listen blocks, serving UDP traffic until the process exits.
func (g *Gateway) Listen() error {
	Log("KCP Gateway is listening at " + g.conn.LocalAddr().String())

	buf := make([]byte, maxPacketSize)
	for {
		n, addr, err := g.conn.ReadFromUDP(buf)
		if err != nil {
			continue
		}
		if n == 20 {
			if err := g.processNetOperation(append([]byte(nil), buf[:n]...), addr); err != nil {
				PacketLog("[ERROR] netop: " + err.Error())
			}
		} else if n >= 28 {
			g.processKCPPayload(append([]byte(nil), buf[:n]...), addr)
		} else {
			PacketLog("unk data len " + itoa(n))
		}
	}
}

func (g *Gateway) processNetOperation(data []byte, addr *net.UDPAddr) error {
	op, err := DecodeNetOperation(data)
	if err != nil {
		return err
	}

	switch {
	case op.Head == NetOpConnectHead && op.Tail == NetOpConnectTail:
		return g.establishSession(op, addr)
	case op.Head == NetOpDropHead && op.Tail == NetOpDropTail:
		g.dropSession(op.Param1, op.Param2, addr)
	default:
		Log("Unknown magic pair received")
	}
	return nil
}

// establishSession mirrors establish_kcp_session: assign conv/token, load
// freesr-data.json, keep only the newest session, and reply 0x145.
func (g *Gateway) establishSession(op *NetOperation, addr *net.UDPAddr) error {
	conv := g.idSeq.Add(1)
	token := randomToken()

	Log("New connection from addr: " + addr.String() + " with conv_id: " + itoa(int(conv)))
	PacketLog("[KCP] New connection from " + addr.String() + " (conv_id: " + itoa(int(conv)) + ", token: " + itoa(int(token)) + ")")

	session := NewSession(func(payload []byte) error {
		_, err := g.conn.WriteToUDP(payload, addr)
		return err
	}, conv, token)

	if data, err := common.LoadFreesrData(); err == nil {
		session.SetJSON(data)
		PacketLog("[KCP] FreesrData loaded successfully")
	} else {
		PacketLog("[ERROR] Failed to load initial freesr-data.json: " + err.Error())
	}

	// file watcher: reload freesr-data.json when the SRTools overwrites it
	go watchFreesrData(session)

	// single-player behaviour from the Rust gateway: drop older sessions
	g.mu.Lock()
	for _, old := range g.sessions {
		old.Close()
	}
	g.sessions = map[uint32]*Session{conv: session}
	g.mu.Unlock()

	// periodic KCP driver (retransmissions, window probes)
	go func() {
		ticker := time.NewTicker(50 * time.Millisecond)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				session.DriveUpdate()
			case <-session.Done():
				return
			}
		}
	}()

	reply := (&NetOperation{
		Head:   NetOpAcceptHead,
		Param1: conv,
		Param2: token,
		Data:   op.Data,
		Tail:   NetOpAcceptTail,
	}).Encode()
	_, err := g.conn.WriteToUDP(reply, addr)
	return err
}

// GetSession returns the live session for a conv id (used by tests/tools).
func (g *Gateway) GetSession(conv uint32) *Session {
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.sessions[conv]
}

func (g *Gateway) dropSession(conv, token uint32, addr *net.UDPAddr) {
	Log("drop_kcp_session " + itoa(int(conv)) + " " + itoa(int(token)))
	g.mu.Lock()
	defer g.mu.Unlock()
	s, ok := g.sessions[conv]
	if !ok {
		Log("drop_kcp_session failed, no session with conv_id " + itoa(int(conv)) + " was found")
		return
	}
	if s.Token == token {
		s.Close()
		delete(g.sessions, conv)
		Log("Client from " + addr.String() + " disconnected")
	}
}

func (g *Gateway) processKCPPayload(data []byte, addr *net.UDPAddr) {
	conv := kcplib.GetConv(data)

	g.mu.Lock()
	session := g.sessions[conv]
	g.mu.Unlock()

	if session == nil {
		Log("Session with conv_id " + itoa(int(conv)) + " not found!")
		return
	}

	// same-conv replays from a stale client get rejected by the token check
	if err := session.Consume(data); err != nil {
		PacketLog("[ERROR] Session consume error (" + addr.String() + "): " + err.Error())
	}
}

// watchFreesrData polls freesr-data.json mtime every second (the Rust version
// used a notify debouncer) and re-syncs the player on change.
func watchFreesrData(session *Session) {
	var lastMod time.Time
	if st, err := statFile("freesr-data.json"); err == nil {
		lastMod = st
	}
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-session.Done():
			return
		case <-ticker.C:
			st, err := statFile("freesr-data.json")
			if err != nil || !st.After(lastMod) {
				continue
			}
			lastMod = st

			if data, err := common.LoadFreesrData(); err == nil {
				session.SetJSON(data)
				if err := session.SyncPlayer(); err == nil {
					Log("json updated")
				}
			}
		}
	}
}