package conn

import (
	"context"
	"io"

	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"

	"github.com/ydb-platform/ydb-go-sdk/v3/internal/conn/gtrace"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/meta"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/operation"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/stack"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
)

type grpcClientStream struct {
	parentConn   *conn
	stream       grpc.ClientStream
	streamCtx    context.Context //nolint:containedctx
	streamCancel context.CancelFunc
	wrapping     bool
	traceID      string
	sentMark     *modificationMark
}

func (s *grpcClientStream) Header() (metadata.MD, error) {
	return s.stream.Header()
}

func (s *grpcClientStream) Trailer() metadata.MD {
	return s.stream.Trailer()
}

func (s *grpcClientStream) Context() context.Context {
	return s.stream.Context()
}

// Endpoint returns the endpoint of the connection used by this stream.
// It implements the optional interface used by topic writer for session logging.
func (s *grpcClientStream) Endpoint() endpoint.Endpoint {
	return s.parentConn.Endpoint()
}

func (s *grpcClientStream) CloseSend() (err error) {
	var (
		ctx    = s.streamCtx
		onDone = gtrace.DriverOnConnStreamCloseSend(s.parentConn.config.Trace(), &ctx,
			stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/v3/internal/conn.(*grpcClientStream).CloseSend"),
		)
	)
	defer func() {
		onDone(err)
	}()

	stop := s.parentConn.lastUsage.Start()
	defer stop()

	err = s.stream.CloseSend()
	if err != nil {
		if !s.wrapping {
			return err
		}

		return xerrors.WithStackTrace(xerrors.Join(
			s.streamCtx.Err(),
			xerrors.Transport(err,
				xerrors.WithAddress(s.parentConn.Address()),
				xerrors.WithNodeID(s.parentConn.NodeID()),
				xerrors.WithTraceID(s.traceID),
			),
		))
	}

	return nil
}

func (s *grpcClientStream) SendMsg(m any) (err error) {
	var (
		ctx    = s.streamCtx
		onDone = gtrace.DriverOnConnStreamSendMsg(s.parentConn.config.Trace(), &ctx,
			stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/v3/internal/conn.(*grpcClientStream).SendMsg"),
		)
	)
	defer func() {
		onDone(err)
	}()

	stop := s.parentConn.lastUsage.Start()
	defer stop()

	err = s.stream.SendMsg(m)
	if err != nil {
		if !s.wrapping {
			return err
		}

		if s.sentMark.canRetry() {
			return xerrors.WithStackTrace(xerrors.Retryable(
				xerrors.Join(
					s.streamCtx.Err(),
					xerrors.Transport(err, xerrors.WithTraceID(s.traceID)),
				),
				xerrors.WithName("SendMsg"),
			))
		}

		return xerrors.WithStackTrace(xerrors.Join(
			s.streamCtx.Err(),
			xerrors.Transport(err,
				xerrors.WithAddress(s.parentConn.Address()),
				xerrors.WithNodeID(s.parentConn.NodeID()),
				xerrors.WithTraceID(s.traceID),
			),
		))
	}

	return nil
}

func (s *grpcClientStream) finish(err error) {
	gtrace.DriverOnConnStreamFinish(s.parentConn.config.Trace(), s.streamCtx,
		stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/v3/internal/conn.(*grpcClientStream).finish"), err,
	)
	s.streamCancel()
}

func (s *grpcClientStream) RecvMsg(m any) (err error) {
	var (
		ctx    = s.streamCtx
		onDone = gtrace.DriverOnConnStreamRecvMsg(s.parentConn.config.Trace(), &ctx,
			stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/v3/internal/conn.(*grpcClientStream).RecvMsg"),
		)
	)
	defer func() {
		onDone(err)
		if err != nil {
			meta.CallTrailerCallback(s.streamCtx, s.stream.Trailer())
		}
	}()

	stop := s.parentConn.lastUsage.Start()
	defer stop()

	err = s.stream.RecvMsg(m)
	if err != nil {
		if xerrors.Is(err, io.EOF) {
			return io.EOF
		}

		if !s.wrapping {
			return err
		}

		if s.sentMark.canRetry() {
			return xerrors.WithStackTrace(xerrors.Retryable(
				xerrors.Join(
					s.streamCtx.Err(),
					xerrors.Transport(err, xerrors.WithTraceID(s.traceID)),
				),
				xerrors.WithName("RecvMsg"),
			))
		}

		return xerrors.WithStackTrace(xerrors.Join(
			s.streamCtx.Err(),
			xerrors.Transport(err,
				xerrors.WithAddress(s.parentConn.Address()),
				xerrors.WithNodeID(s.parentConn.NodeID()),
				xerrors.WithTraceID(s.traceID),
			),
		))
	}

	if s.wrapping {
		if operation, ok := m.(operation.Status); ok {
			if status := operation.GetStatus(); status != Ydb.StatusIds_SUCCESS {
				return xerrors.WithStackTrace(xerrors.Operation(
					xerrors.FromOperation(operation),
					xerrors.WithAddress(s.parentConn.Address()),
					xerrors.WithNodeID(s.parentConn.NodeID()),
				))
			}
		}
	}

	return nil
}
