You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: internal.go
    path of file: ./repos/streamdal/apps/server/apis/grpcapi
    
    the code of the file until where you have to start completion: package grpcapi

import (
	"context"
	"fmt"
	"io"
	"strings"
	"time"

	"github.com/cactus/go-statsd-client/v5/statsd"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	"github.com/streamdal/streamdal/libs/protos/build/go/protos"
	"github.com/streamdal/streamdal/libs/protos/build/go/protos/shared"

	"github.com/streamdal/streamdal/apps/server/types"
	"github.com/streamdal/streamdal/apps/server/util"
	"github.com/streamdal/streamdal/apps/server/validate"
)

const (
	MaxKVCommandSizeBytes = 64 * 1024 // 64KB
)

// InternalServer implements the internal GRPC API interface
type InternalServer struct {
	GRPCAPI
	// Must be implemented in order to satisfy the protos InternalServer interface
	protos.UnimplementedInternalServer
}

func (g *GRPCAPI) newInternalServer() *InternalServer {
	return &InternalServer{
		GRPCAPI: *g,
	}
}

// sendActiveTails is executed during Register() and will send all active tails
// to the client (so that SDK can "resume" any in-progress tails)
func (s *InternalServer) sendActiveTails(ctx context.Co
    
func (s *InternalServer) SendTail(srv protos.Internal_SendTailServer) error {
	llog := s.log.WithFields(logrus.Fields{
		"method":     "SendTail",
		"request_id": util.CtxRequestId(srv.Context()),
	})

	// This isn't necessary for go, but other language libraries, such as python
	// require a response to eventually be sent and will throw an exception if
	// one is not received
	defer func() {
		llog.Debug("sending final tail response")

		srv.SendAndClose(&protos.StandardResponse{
			Id:   util.CtxRequestId(srv.Context()),
			Code: protos.ResponseCode_RESPONSE_CODE_OK,
		})
	}()

	for {
		select {
		case <-srv.Context().Done():
			llog.Debug("detected client disconnect")
			return nil
		case <-s.Options.ShutdownContext.Done():
			llog.Debug("server shutting down, exiting stream")
			return nil
		default:
			tailResp, err := srv.Recv()
			if err != nil {
				if strings.Contains(err.Error(), io.EOF.Error()) || strings.Contains(err.Error(), context.Canceled.Error()) {
					llog.Debug("client closed stream")
					return nil
				}

				llog.Error(errors.Wrap(err, "unable to receive tail response"))
				continue
			}

			llog.Debugf("after srv.Recv(), before validation for session id '%s', tail request id '%s'",
				tailResp.SessionId, tailResp.TailRequestId)

			if err := validate.TailResponse(tailResp); err != nil {
				llog.Error(errors.Wrapf(err, "invalid tail response for session id '%s', tail request id '%s'",
					tailResp.SessionId, tailResp.TailRequestId))
				continue
			}

			llog.Debugf("after validation, before BroadcastTailResponse() for session id '%s', tail request id '%s'",
				tailResp.SessionId, tailResp.TailRequestId)

			if err := s.Options.BusService.BroadcastTailResponse(srv.Context(), tailResp); err != nil {
				llog.Error(errors.Wrapf(err, "unable to broadcast tail response for session id '%s', tail request id '%s'",
					tailResp.SessionId, tailResp.TailRequestId))

				continue
			}

			llog.Debugf("after BroadcastTailResponse() for session id '%s', tail request id '%s'",
				tailResp.SessionId, tailResp.TailRequestId)
		}
	}
}