path: root/internal/server/server.go

package server

import (
	"context"
	"crypto/rand"
	"crypto/tls"
	_ "embed"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/kyren223/eko/internal/data"
	"github.com/kyren223/eko/internal/packet"
	"github.com/kyren223/eko/pkg/assert"
	"github.com/kyren223/eko/pkg/snowflake"
)

//go:embed server.crt
var certPEM []byte

//go:embed server.key
var keyPEM []byte

var (
	nodeId    int64 = 0
	tlsConfig *tls.Config

	unsupportedEncodingErrorPacket packet.Packet
	unsupportedTypeErrorPacket     packet.Packet
)

var ErrClosedNilListener error = errors.New("server: close on nil listener")

func init() {
	cert, err := tls.X509KeyPair(certPEM, keyPEM)
	if err != nil {
		log.Fatalln("error loading certificate:", err)
	}

	tlsConfig = &tls.Config{
		Certificates: []tls.Certificate{cert},
	}

	message := packet.ErrorMessage{Error: packet.PacketUnsupportedEncoding.Error()}
	encoder, err := packet.NewMsgPackEncoder(&message)
	assert.NoError(err, "constant packets should not error")
	unsupportedEncodingErrorPacket = packet.NewPacket(encoder)

	message = packet.ErrorMessage{Error: packet.PacketUnsupportedType.Error()}
	encoder, err = packet.NewMsgPackEncoder(&message)
	assert.NoError(err, "constant packets should not error")
	unsupportedTypeErrorPacket = packet.NewPacket(encoder)
}

type server struct {
	node *snowflake.Node
	port uint16
}

func NewServer(port uint16) server {
	assert.Assert(nodeId <= snowflake.NodeMax, "maximum amount of servers reached %v", snowflake.NodeMax)
	node := snowflake.NewNode(nodeId)
	nodeId++

	return server{
		node: node,
		port: port,
	}
}

func (s *server) ListenAndServe(ctx context.Context) {
	listener, err := tls.Listen("tcp4", ":"+strconv.Itoa(int(s.port)), tlsConfig)
	if err != nil {
		log.Fatalf("error starting server: %s", err)
	}
	defer listener.Close()
	go func() {
		<-ctx.Done()
		listener.Close()
	}()

	log.Printf("started listening on port %v...\n", s.port)
	var wg sync.WaitGroup
	for {
		conn, err := listener.Accept()
		if err != nil {
			if !errors.Is(err, net.ErrClosed) {
				log.Println("error accepting connection:", err)
			}
			break
		}
		wg.Add(1)
		go func() {
			handleConnection(ctx, conn)
			wg.Done()
		}()
	}
	log.Printf("stopped listening on port %v\n", s.port)

	log.Println("waiting for all active connections to close...")
	wg.Wait()
	log.Println("server shutdown complete")
}

func handleConnection(ctx context.Context, conn net.Conn) {
	addr, ok := conn.RemoteAddr().(*net.TCPAddr)
	assert.Assert(ok, "getting tcp address should never fail as we are using tcp connections")

	writeQueue := make(chan packet.Packet, 10)
	session := newSession(addr, writeQueue)
	nonce := session.Challenge()

	ctx = newContext(ctx, session)
	framer := packet.NewFramer(ctx)

	log.Println(addr, "accepted")

	defer func() {
		conn.Close()
		log.Println(addr, "disconnected")
	}()

	go func() {
		var mu sync.Mutex
		for {
			packet, ok := <-writeQueue
			if !ok {
				break
			}
			mu.Lock()
			packet.Into(conn)
			mu.Unlock()
		}
	}()

	go func() {
		for {
			request, ok := <-framer.Out
			if !ok {
				break
			}
			response := processPacket(ctx, request)
			writeQueue <- response
		}
	}()

	buffer := make([]byte, 512)
	for {
		n, err := conn.Read(buffer)
		if err != nil {
			if !errors.Is(err, io.EOF) {
				log.Println(addr, "read error:", err)
			}
			break
		}

		err = framer.Push(ctx, buffer[:n])
		if err != nil {
			// Wrap err and send to client then break
		}
	}
}

func _handleConnection(ctx context.Context, conn net.Conn) {
	addr, ok := conn.RemoteAddr().(*net.TCPAddr)
	assert.Assert(ok, "getting tcp address should be valid")

	log.Println(addr, "accepted")
	defer log.Println(addr, "disconnected")
	defer conn.Close()

	// ctx = newContext(ctx, addr)
	// // TODO: consider adding timeout/deadline to ctx?

	out, outErr := packet.RunFramer(ctx, conn)
outer:
	for {
		select {
		case packet, ok := <-out:
			if !ok {
				break outer
			}
			log.Printf("client %v: request packet: %v\n", conn.RemoteAddr().String(), packet)
			responsePacket, err := handlePacket(packet)
			log.Printf("client %v: response packet: %v\n", conn.RemoteAddr().String(), responsePacket)
			if err != nil {
				log.Printf("client %v: error processing request: %v\n", conn.RemoteAddr().String(), err)
				break outer
			}
			_, err = responsePacket.Into(conn)
			if err != nil {
				log.Printf("client %v: error writing packet: %v\n", conn.RemoteAddr().String(), err)
				break outer
			}

		case err := <-outErr:
			if err == packet.PacketUnsupportedEncoding {
				_, err := unsupportedEncodingErrorPacket.Into(conn)
				log.Printf("client %v: error writing unsupported encoding packet: %v\n", conn.RemoteAddr().String(), err)
			} else if err == packet.PacketUnsupportedType {
				_, err := unsupportedTypeErrorPacket.Into(conn)
				log.Printf("client %v: error writing unsupported type packet: %v\n", conn.RemoteAddr().String(), err)
			} else if err != nil {
				log.Printf("client %v: internal error: %v\n", conn.RemoteAddr().String(), err)
			}
			break outer

		case <-ctx.Done():
			log.Printf("client %v: %v\n", conn.RemoteAddr().String(), ctx.Err())
			break outer
		}
	}
}

type Session struct {
	Addr       *net.TCPAddr
	WriteQueue <-chan packet.Packet

	mu         sync.Mutex
	challenge   []byte
	issuedTime time.Time
}

func newSession(addr *net.TCPAddr, writeQueue <-chan packet.Packet) *Session {
	session := &Session{
		Addr:       addr,
		WriteQueue: writeQueue,
		challenge:  make([]byte, 32), // Recommended nonce size
	}
	session.Challenge() // Make sure an initial nonce is generated
	return session
}

func (s *Session) Challenge() []byte {
	s.mu.Lock()
	defer s.mu.Unlock()
	if time.Since(s.issuedTime) > time.Minute {
		s.issuedTime = time.Now()
		_, err := rand.Read(s.challenge)
		assert.NoError(err, "random should always produce a value")
	}
	return s.challenge
}

type key struct{}

var sessKey key

func newContext(ctx context.Context, sess *Session) context.Context {
	return context.WithValue(ctx, sessKey, sess)
}

func FromContext(ctx context.Context) (*Session, bool) {
	sess, ok := ctx.Value(sessKey).(*Session)
	return sess, ok
}

// TODO: Move everything below this to somewhere else

func handlePacket(pkt packet.Packet) (packet.Packet, error) {
	var response packet.TypedMessage
	switch pkt.Type() {
	case packet.PacketEko:
		var request packet.EkoMessage
		if err := pkt.DecodePayload(&request); err != nil {
			return packet.Packet{}, fmt.Errorf("decode error: %v", err)
		}

		response = &packet.EkoMessage{Message: "Eko \"" + request.Message + "\""}
	case packet.PacketSendMessage:
		var request packet.SendMessageMessage
		if err := pkt.DecodePayload(&request); err != nil {
			return packet.Packet{}, fmt.Errorf("decode error: %v", err)
		}

		content := strings.TrimSpace(request.Content)
		if content == "" {
			response = &packet.ErrorMessage{Error: "content must not be blank"}
			break
		}

		message := data.Message{
			Id:          node.Generate(),
			SenderId:    node.Generate(),
			FrequencyId: node.Generate(),
			NetworkId:   node.Generate(),
			Contents:    content,
		}
		messages = append(messages, message)

		response = &packet.EkoMessage{Message: "Eko OK"}
	case packet.PacketGetMessages:
		var request packet.GetMessagesMessage
		if err := pkt.DecodePayload(&request); err != nil {
			return packet.Packet{}, fmt.Errorf("decode error: %v", err)
		}

		response = &packet.MessagesMessage{Messages: messages}
	default:
		return packet.Packet{}, errors.New("TODO: not implemented yet")
	}

	assert.NotNil(response, "response must always be set")
	encoder, err := packet.NewMsgPackEncoder(response)
	if err != nil {
		return packet.Packet{}, fmt.Errorf("encode error: %v", err)
	}
	return packet.NewPacket(encoder), nil
}

var (
	node     = snowflake.NewNode(1)
	messages []data.Message
)