package scdbclient

import (
	"context"
	"errors"

	"github.com/sjy-dv/scdb/scdb/pkg/log"
	"github.com/sjy-dv/scdb/scdb/pkg/protobuf/scdbpb"
)

type transaction struct {
	tx scdbpb.Scdb_TransactionClient
	c  *connectionpool
}

// Begin SCDB transaction, return nil is begin error
func (proxy *proxyClient) Begin() *transaction {
	conn := proxy.getPool()
	if conn == nil {
		return nil
	}
	stream, err := wrapper(conn.connection).Transaction(context.Background())
	if err != nil {
		proxy.inValid(conn)
		conn.sessions.Add(-1)
		return nil
	}
	err = stream.Send(&scdbpb.TxIn{
		Begin: true,
	})
	if err != nil {
		return nil
	}
	response, err := stream.Recv()
	if err != nil {
		proxy.inValid(conn)
		conn.sessions.Add(-1)
		return nil
	}
	if response.ErrorStatus {
		conn.sessions.Add(-1)
		return nil
	}
	return &transaction{
		tx: stream,
		c:  conn,
	}
}

// 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 (t *transaction) Get(ctx context.Context, key string) ([]byte, error) {
	if err := t.tx.Send(&scdbpb.TxIn{
		Do:  "GET",
		Key: []byte(key),
	}); err != nil {
		return nil, err
	}
	response, err := t.tx.Recv()
	if err != nil {
		return nil, err
	}
	if response.ErrorStatus {
		return nil, errors.New(response.Error)
	}
	return response.GetValue(), nil
}

// Stores a value using a key and value. If the storage fails, it returns an error, and if successful, it returns nil.
func (t *transaction) Save(ctx context.Context, key string, value []byte) error {
	if err := t.tx.Send(&scdbpb.TxIn{
		Do:    "SET",
		Key:   []byte(key),
		Value: value,
	}); err != nil {
		return err
	}
	response, err := t.tx.Recv()
	if err != nil {
		return err
	}
	if response.ErrorStatus {
		return errors.New(response.Error)
	}
	return nil
}

// Deletes data based on the key. If successful, it returns nil, and if it fails, it returns an error.
func (t *transaction) Del(ctx context.Context, key string) error {
	if err := t.tx.Send(&scdbpb.TxIn{
		Do:  "DEL",
		Key: []byte(key),
	}); err != nil {
		return err
	}
	response, err := t.tx.Recv()
	if err != nil {
		return err
	}
	if response.ErrorStatus {
		return errors.New(response.Error)
	}
	return nil
}

// Commit SCDB transaction
func (t *transaction) Commit() error {
	defer t.tx.CloseSend()
	defer t.c.sessions.Add(-1)
	if err := t.tx.Send(&scdbpb.TxIn{
		Commit: true,
	}); err != nil {
		return err
	}
	response, err := t.tx.Recv()
	if err != nil {
		return err
	}
	if response.ErrorStatus {
		return errors.New(response.Error)
	}
	return nil
}

// Rollback SCDB transaction
func (t *transaction) Rollback() {
	defer t.tx.CloseSend()
	defer t.c.sessions.Add(-1)
	if err := t.tx.Send(&scdbpb.TxIn{
		Rollback: true,
	}); err != nil {
		log.Errorf("scdb transaction rollback error: %v", err)
	}
	response, err := t.tx.Recv()
	if err != nil {
		log.Errorf("scdb transaction rollback error: %v", err)
	}
	if response.ErrorStatus {
		log.Errorf("scdb transaction rollback error: %s", response.ErrorStatus)
	}
}
