package rpc

import (
	"context"
	"encoding/binary"
	"errors"
	"gitee.com/mosache/YFrame/micro/rpc/compress"
	"gitee.com/mosache/YFrame/micro/rpc/compress/nothing"
	"gitee.com/mosache/YFrame/micro/rpc/internal/errs"
	"gitee.com/mosache/YFrame/micro/rpc/message"
	"gitee.com/mosache/YFrame/micro/rpc/serialize"
	"gitee.com/mosache/YFrame/micro/rpc/serialize/json"
	"net"
	"reflect"
	"strconv"
	"time"
)

type Server struct {
	services    map[string]reflectStub
	serializers map[uint8]serialize.Serializer
	compressors map[uint8]compress.Compressor
}

type reflectStub struct {
	s           Service
	value       reflect.Value
	serializers map[uint8]serialize.Serializer
	compressors map[uint8]compress.Compressor
}

func (s reflectStub) invoke(ctx context.Context, req *message.Request) ([]byte, error) {
	var (
		serializer serialize.Serializer
		compressor compress.Compressor
		err        error
		ok         bool
	)
	method := s.value.MethodByName(req.MethodName)

	inReq := reflect.New(method.Type().In(1).Elem()).Interface()

	if serializer, ok = s.serializers[req.Serializer]; !ok {
		return nil, errs.ErrServerSerializerNotFound
	}
	if compressor, ok = s.compressors[req.Compresser]; !ok {
		return nil, errs.ErrServerCompressorNotFound
	}

	req.Data, err = compressor.UnCompress(req.Data)
	if err != nil {
		return nil, err
	}

	err = serializer.Decode(req.Data, inReq)
	if err != nil {
		return nil, err
	}

	results := method.Call([]reflect.Value{reflect.ValueOf(ctx), reflect.ValueOf(inReq)})
	if err := results[1].Interface(); err != nil {
		return nil, results[1].Interface().(error)
	}

	var (
		res []byte
		er  error
	)
	res, er = serializer.Encode(results[0].Interface())
	if er != nil {
		return nil, er
	}
	res, er = compressor.Compress(res)
	if er != nil {
		return nil, er
	}

	return res, nil
}

func NewServer() *Server {

	s := &Server{
		services:    make(map[string]reflectStub),
		serializers: map[uint8]serialize.Serializer{},
		compressors: map[uint8]compress.Compressor{},
	}

	s.RegisterSerializer(json.Serializer{})
	s.RegisterCompressor(nothing.Compressor{})
	return s

}

func (s *Server) Register(srv Service) {
	s.services[srv.Name()] = reflectStub{
		s:           srv,
		value:       reflect.ValueOf(srv),
		serializers: s.serializers,
		compressors: s.compressors,
	}
}

func (s *Server) RegisterSerializer(serializer serialize.Serializer) {
	s.serializers[serializer.Code()] = serializer
}

func (s *Server) RegisterCompressor(compressor compress.Compressor) {
	s.compressors[compressor.Code()] = compressor
}

func (s *Server) Start(addr string) error {

	listener, err := net.Listen("tcp", addr)
	if err != nil {
		return err
	}
	defer func() { _ = listener.Close() }()
	for {
		conn, err := listener.Accept()
		if err != nil {
			continue
		}

		go func() {
			if err := s.handConn(conn); err != nil {
				conn.Close()
			}
		}()
	}
}

func (s *Server) handConn(conn net.Conn) error {
	dataBs, err := ReadMsg(conn)

	req := message.DecodeReq(dataBs)

	ctx := context.Background()
	cancel := func() {}
	if deadlineStr, ok := req.Meta["deadline"]; ok {
		if deadline, er := strconv.ParseInt(deadlineStr, 10, 64); er == nil {
			ctx, cancel = context.WithDeadline(ctx, time.UnixMilli(deadline))
		}
	}

	if oneWay, ok := req.Meta["onWay"]; ok {
		if oneWay == "onWay" {
			ctx = ContextWithOneway(ctx)
		}
	}

	var resp *message.Response
	if resp, err = s.Invoke(ctx, req); err != nil {
		/// 业务错误
		resp.Error = []byte(err.Error())
	}

	cancel()

	resp.CalculateHeaderLength()
	resp.CalculateBodyLength()

	/// 写回响应
	if _, err = conn.Write(message.EncodeResp(resp)); err != nil {
		return err
	}
	//if err = s.send(conn, message.EncodeResp(resp)); err != nil {
	//	return err
	//}

	return nil
}

//func (s *Server) handMsg(bs []byte) (any, error) {
//
//	req := message.DecodeReq(bs)
//
//	ret, err := s.Invoke(context.Background(), req)
//
//	/// biz err
//	if err != nil {
//		return nil, err
//	}
//
//	return ret, nil
//}

func (s *Server) Invoke(ctx context.Context, req *message.Request) (*message.Response, error) {
	srv, ok := s.services[req.ServiceName]

	resp := &message.Response{
		RequestID:  req.RequestID,
		Version:    req.Version,
		Compresser: req.Compresser,
		Serializer: req.Serializer,
	}

	defer func() {
		resp.CalculateHeaderLength()
		resp.CalculateBodyLength()
	}()

	if !ok {
		return nil, errors.New("[rpc] service not found")
	}

	if IsOneWayContext(ctx) {
		go func() {
			_, _ = srv.invoke(ctx, req)
		}()

		return resp, errs.ErrServerOneWayCall
	}

	respData, err := srv.invoke(ctx, req)
	if err != nil {
		return nil, err
	}

	resp.Data = respData

	return resp, nil
}

func (s *Server) send(conn net.Conn, data []byte) error {
	dataLen := len(data)

	dataBs := make([]byte, dataLen+sendDataHeaderBytes)

	binary.BigEndian.PutUint64(dataBs[:sendDataHeaderBytes], uint64(dataLen))

	copy(dataBs[sendDataHeaderBytes:], data)

	_, err := conn.Write(dataBs)
	if err != nil {
		return err
	}

	return nil
}