package kit_test

import (
	"bytes"
	"context"
	"io"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/clubpay/ronykit/kit"
	"github.com/clubpay/ronykit/kit/desc"
	"github.com/clubpay/ronykit/kit/utils"
)

type testSelector struct{}

func (t testSelector) Query(_ string) any {
	return nil
}

func (t testSelector) GetEncoding() kit.Encoding {
	return kit.JSON
}

func (t testSelector) String() string {
	return "testSelector"
}

type testGateway struct {
	d kit.GatewayDelegate
}

var _ kit.Gateway = (*testGateway)(nil)

func (t *testGateway) Send(c *testConn, msg []byte) {
	t.d.OnMessage(c, msg)
}

func (t *testGateway) Start(_ context.Context, _ kit.GatewayStartConfig) error {
	return nil
}

func (t *testGateway) Shutdown(_ context.Context) error {
	return nil
}

func (t *testGateway) Subscribe(d kit.GatewayDelegate) {
	t.d = d
}

func (t *testGateway) Dispatch(ctx *kit.Context, in []byte) (kit.ExecuteArg, error) {
	ctx.In().SetMsg(kit.RawMessage(in))

	return kit.ExecuteArg{
		ServiceName: "testService",
		ContractID:  "testService.1",
		Route:       "someRoute",
	}, nil
}

func (t *testGateway) Register(
	_, _ string, _ kit.Encoding, _ kit.RouteSelector, _, _ kit.Message,
) {
}

type testCluster struct {
	mtx       sync.Mutex
	delegates map[string]kit.ClusterDelegate
	kv        map[string]string
	m         chan struct {
		id   string
		data []byte
	}
}

var (
	_ kit.Cluster      = (*testCluster)(nil)
	_ kit.ClusterStore = (*testCluster)(nil)
)

func newTestCluster() *testCluster {
	t := &testCluster{
		delegates: map[string]kit.ClusterDelegate{},
		m: make(chan struct {
			id   string
			data []byte
		}, 10),
	}

	go func() {
		for x := range t.m {
			t.mtx.Lock()
			d, ok := t.delegates[x.id]
			t.mtx.Unlock()
			if ok {

				d.OnMessage(x.data)
			}
		}
	}()

	return t
}

func (t *testCluster) Start(ctx context.Context) error {
	return nil
}

func (t *testCluster) Shutdown(ctx context.Context) error {
	return nil
}

func (t *testCluster) Subscribe(id string, d kit.ClusterDelegate) {
	t.mtx.Lock()
	if t.delegates == nil {
		t.delegates = map[string]kit.ClusterDelegate{}
	}
	t.delegates[id] = d
	t.mtx.Unlock()
}

func (t *testCluster) Subscribers() ([]string, error) {
	var members []string
	t.mtx.Lock()
	for m := range t.delegates {
		members = append(members, m)
	}
	t.mtx.Unlock()

	return members, nil
}

func (t *testCluster) Publish(id string, data []byte) error {
	t.m <- struct {
		id   string
		data []byte
	}{id: id, data: data}

	return nil
}

func (t *testCluster) Store() kit.ClusterStore {
	return t
}

func (t *testCluster) Set(ctx context.Context, key, value string, _ time.Duration) error {
	if t.kv == nil {
		t.kv = map[string]string{}
	}

	t.kv[key] = value

	return nil
}

func (t *testCluster) SetMulti(ctx context.Context, kv map[string]string, ttl time.Duration) error {
	if t.kv == nil {
		t.kv = map[string]string{}
	}

	for k, v := range kv {
		t.kv[k] = v
	}

	return nil
}

func (t *testCluster) Delete(ctx context.Context, key string) error {
	if t.kv == nil {
		t.kv = map[string]string{}
	}

	delete(t.kv, key)

	return nil
}

func (t *testCluster) Get(ctx context.Context, key string) (string, error) {
	return t.kv[key], nil
}

func (t *testCluster) Scan(ctx context.Context, prefix string, cb func(key string) bool) error {
	for k := range t.kv {
		if strings.HasPrefix(k, prefix) {
			if !cb(k) {
				return nil
			}
		}
	}

	return nil
}

func (t *testCluster) ScanWithValue(ctx context.Context, prefix string, cb func(string, string) bool) error {
	for k, v := range t.kv {
		if strings.HasPrefix(k, prefix) {
			if !cb(k, v) {
				return nil
			}
		}
	}

	return nil
}

type testConn struct {
	id       uint64
	clientIP string
	stream   bool
	kv       map[string]string
	buf      *bytes.Buffer
}

var _ kit.Conn = (*testConn)(nil)

func newTestConn(id uint64, clientIP string, stream bool) *testConn {
	return &testConn{
		id:       id,
		clientIP: clientIP,
		stream:   stream,
		kv:       map[string]string{},
		buf:      &bytes.Buffer{},
	}
}

func (t testConn) ConnID() uint64 {
	return t.id
}

func (t testConn) ClientIP() string {
	return t.clientIP
}

func (t testConn) Write(data []byte) (int, error) {
	return t.buf.Write(data)
}

func (t *testConn) WriteEnvelope(e *kit.Envelope) error {
	b, err := kit.MarshalMessage(e.GetMsg())
	if err != nil {
		return err
	}
	_, err = t.buf.Write(b)

	return err
}

func (t testConn) Read() ([]byte, error) {
	return io.ReadAll(t.buf)
}

func (t testConn) ReadString() (string, error) {
	d, err := t.Read()
	if err != nil {
		return "", err
	}
	return string(d), nil
}

func (t testConn) Stream() bool {
	return t.stream
}

func (t testConn) Walk(f func(key string, val string) bool) {
	for k, v := range t.kv {
		if !f(k, v) {
			return
		}
	}
}

func (t testConn) Get(key string) string {
	return t.kv[key]
}

func (t *testConn) Set(key string, val string) {
	t.kv[key] = val
}

func (t testConn) Keys() []string {
	keys := make([]string, 0, len(t.kv))
	for k := range t.kv {
		keys = append(keys, k)
	}

	return keys
}

// testRESTSelector implements kit.RESTRouteSelector for testing purposes.
type testRESTSelector struct {
	enc    kit.Encoding
	method string
	path   string
}

func (t testRESTSelector) Query(q string) any {
	return nil
}

func (t testRESTSelector) GetEncoding() kit.Encoding {
	return t.enc
}

func (t testRESTSelector) GetMethod() string {
	return t.method
}

func (t testRESTSelector) GetPath() string {
	return t.path
}

// testRPCSelector implements kit.RPCSelector for testing purposes.
type testRPCSelector struct {
	enc       kit.Encoding
	predicate string
}

func (t testRPCSelector) Query(_ string) any {
	return nil
}

func (t testRPCSelector) GetEncoding() kit.Encoding {
	return t.enc
}

func (t testRPCSelector) GetPredicate() string {
	return t.predicate
}

func BenchmarkServer(b *testing.B) {
	bundle := &testGateway{}
	s := kit.NewServer(
		kit.WithGateway(bundle),
		kit.WithService(
			desc.NewService("testService").
				AddContract(
					desc.NewContract().
						AddRoute(desc.Route("", testSelector{})).
						AddHandler(
							func(ctx *kit.Context) {
								ctx.Out().
									SetMsg(ctx.In().GetMsg()).
									Send()

								return
							},
						),
				).
				Build(),
		),
	).Start(b.Context())
	defer s.Shutdown(b.Context())

	req := []byte(utils.RandomID(24))
	b.ResetTimer()
	b.ReportAllocs()
	b.RunParallel(
		func(pb *testing.PB) {
			for pb.Next() {
				c := newTestConn(utils.RandomUint64(0), "", false)
				bundle.Send(c, req)
			}
		},
	)
}
