pkg/plug/grpc.go

@@ -157,10 +157,54 @@ func (s *server) removeTask(taskID string) {
 }
 
 func (s *server) GetUserMessages(ctx context.Context, message *gRPC.GetMessagesRequest) (*gRPC.GetMessagesResponse, error) {
+	ctx, span := tracer.Start(ctx, "GetUserMessages")
+	defer span.End()
+
+	sourceID := models.UserSourceID(message.UserSourceId)
+	userSource := models.UserSource{BaseModel: models.BaseModel[models.UserSourceID]{ID: sourceID}}
+
+	messages, err := s.getMessageExecution(ctx, userSource)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return nil, err
+	}
+
+	var response gRPC.GetMessagesResponse
+	for _, message := range messages {
+		response.Messages = append(response.Messages, &gRPC.Message{
+			FromUserSourceId: string(userSource.ID),
+			CreatedAt:        message.CreatedAt,
+			Body:             message.Body,
+			Title:            message.Title,
+		})
+	}
+
+	return &response, err
 
 }
 
 func (s *server) SendMessage(ctx context.Context, message *gRPC.SendMessageRequest) (*gRPC.SendMessageResponse, error) {
+	ctx, span := tracer.Start(ctx, "SendMessage")
+	defer span.End()
+
+	response := &gRPC.SendMessageResponse{
+		Success: false,
+	}
+
+	sourceID := models.UserSourceID(message.UserSourceId)
+	userSource := models.UserSource{BaseModel: models.BaseModel[models.UserSourceID]{ID: sourceID}}
+
+	err := s.sendMessageExecution(ctx, userSource, message.Message)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+
+		return response, err
+	}
+
+	response.Success = true
+	return response, err
 }
 
 func (s *server) Ping(ctx context.Context, ping *gRPC.PingRequest) (*gRPC.PongResponse, error) {