Messaging/pkg/internal/grpc/stream.go

package grpc

import (
	"context"
	"fmt"

	"git.solsynth.dev/hydrogen/dealer/pkg/hyper"
	"git.solsynth.dev/hydrogen/dealer/pkg/proto"
	"git.solsynth.dev/hydrogen/messaging/pkg/internal/gap"
	"git.solsynth.dev/hydrogen/messaging/pkg/internal/server/exts"
	"git.solsynth.dev/hydrogen/messaging/pkg/internal/services"
	jsoniter "github.com/json-iterator/go"
)

func (v *Server) EmitStreamEvent(_ context.Context, in *proto.StreamEventRequest) (*proto.StreamEventResponse, error) {
	sc := proto.NewStreamControllerClient(gap.H.GetDealerGrpcConn())

	switch in.GetEvent() {
	case "status.typing":
		var data struct {
			ChannelID uint `json:"channel_id" validate:"required"`
		}

		err := jsoniter.Unmarshal(in.GetPayload(), &data)
		if err == nil {
			err = exts.ValidateStruct(data)
		}
		if err != nil {
			_, _ = sc.PushStream(context.Background(), &proto.PushStreamRequest{
				ClientId: &in.ClientId,
				Body: hyper.NetworkPackage{
					Action:  "error",
					Message: fmt.Sprintf("unable parse payload: %v", err),
				}.Marshal(),
			})
		}

		err = services.SetTypingStatus(data.ChannelID, uint(in.GetUserId()))
		if err != nil {
			_, _ = sc.PushStream(context.Background(), &proto.PushStreamRequest{
				ClientId: &in.ClientId,
				Body: hyper.NetworkPackage{
					Action:  "error",
					Message: fmt.Sprintf("unable boardcast status: %v", err),
				}.Marshal(),
			})
		}
	}

	return &proto.StreamEventResponse{}, nil
}