package scdbclient

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/sjy-dv/scdb/scdb/pkg/protobuf/scdbpb"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/types/known/emptypb"
)

// A single gRPC channel can hold 100 sessions, which means it is capable of handling 100 processes simultaneously.
func CreateSessionPool(opt *Options) *proxyClient {
	gRpcClientPool := &gRpcClientPool{}
	gRpcClientPool.addr = address(opt.Host, opt.Port)
	gRpcClientPool.pool = make([]*connectionpool, opt.MingRpcChannl)
	gRpcClientPool.minchannelsize = opt.MingRpcChannl
	gRpcClientPool.maxchannelsize = opt.MaxgRpcChannel
	for i := 0; i < int(opt.MingRpcChannl); i++ {
		status := true
		conn, err := grpc.Dial(gRpcClientPool.addr,
			grpc.WithTransportCredentials(insecure.NewCredentials()))
		if err != nil {
			conn = nil
			status = false
		}
		gRpcClientPool.pool[i] = &connectionpool{
			connection: conn,
			lastCall:   time.Now(),
			status:     status,
			sessions:   &atomic.Int32{},
		}
	}
	return &proxyClient{gRpcClientPool}
}

func wrapper(c *grpc.ClientConn) scdbpb.ScdbClient {
	return scdbpb.NewScdbClient(c)
}

func address(host string, port int) string {
	return fmt.Sprintf("%s:%d", host, port)
}

// Ping-Pong connection, when the connection is established returns nil. but when the connection is closed returns error
func (proxy *proxyClient) Ping(ctx context.Context) error {
	conn := proxy.getPool()
	defer conn.sessions.Add(-1)
	_, err := wrapper(conn.connection).Ping(ctx, &emptypb.Empty{})
	if err != nil {
		proxy.inValid(conn)
		return err
	}
	return nil
}

// Stores a value using a key and value. If the storage fails, it returns an error, and if successful, it returns nil.
func (proxy *proxyClient) Save(ctx context.Context, key string, value []byte) error {
	conn := proxy.getPool()
	defer conn.sessions.Add(-1)
	retval, err := wrapper(conn.connection).Save(ctx, &scdbpb.KV{Key: []byte(key), Value: value})
	if err != nil {
		proxy.inValid(conn)
		return err
	}
	if retval.ErrorStatus {
		return errors.New(retval.Error)
	}
	return nil
}

// Based on the key, it retrieves the value. If the retrieval fails, it returns an error, and if successful, it returns the value associated with the key.
func (proxy *proxyClient) Get(ctx context.Context, key string) ([]byte, error) {
	conn := proxy.getPool()
	defer conn.sessions.Add(-1)
	retval, err := wrapper(conn.connection).Get(ctx, &scdbpb.KV{Key: []byte(key)})
	if err != nil {
		proxy.inValid(conn)
		return nil, err
	}
	if retval.ErrorStatus {
		return nil, errors.New(retval.Error)
	}
	return retval.GetValue(), nil
}

// Deletes data based on the key. If successful, it returns nil, and if it fails, it returns an error.
func (proxy *proxyClient) Del(ctx context.Context, key string) error {
	conn := proxy.getPool()
	defer conn.sessions.Add(-1)
	retval, err := wrapper(conn.connection).Del(ctx, &scdbpb.KV{Key: []byte(key)})
	if err != nil {
		proxy.inValid(conn)
		return err
	}
	if retval.ErrorStatus {
		return errors.New(retval.Error)
	}
	return nil
}

func (proxy *proxyClient) inValid(cp *connectionpool) {
	ctx, canel := context.WithTimeout(context.Background(), time.Second*3)
	defer canel()
	_, err := wrapper(cp.connection).Ping(ctx, &emptypb.Empty{})
	if err != nil {
		proxy.removeConnection(cp)
	}
}

// Closes the connection pool.
func (proxy *proxyClient) Close() error {
	errChan := make(chan error)
	wg := new(sync.WaitGroup)
	for index := range proxy.pool {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			if err := proxy.pool[i].connection.Close(); err != nil {
				errChan <- err
			}
			proxy.pool[i] = nil
		}(index)
	}

	go func() {
		wg.Wait()
		close(errChan)
	}()

	for e := range errChan {
		if e != nil {
			return e
		}
	}
	return nil
}
