ChatGpt to message

src/constant/config/service.go

@@ -35,3 +35,4 @@ const VideoProcessorRpcServiceName = "GuGoTik-VideoProcessorService"
 
 const VideoPicker = "GuGoTik-VideoPicker"
 const Event = "GuGoTik-Recommend"
+const MsgConsumer = "GuGoTik-MgsConsumer"

src/constant/strings/service.go

@@ -13,5 +13,6 @@ const (
 	VideoCommentEvent   = "video.comment.action"
 	VideoPublishEvent   = "video.publish.action"
 
-	MessageActionEvent = "message.send"
+	MessageActionEvent    = "message.send"
+	MessageGptActionEvent = "message.gpt.send"
 )

src/services/message/handler.go

@@ -12,14 +12,14 @@ import (
 	"GuGoTik/src/rpc/user"
 	"GuGoTik/src/storage/database"
 	"GuGoTik/src/storage/redis"
-	grpc2 "GuGoTik/src/utils/grpc"
 	"GuGoTik/src/utils/logging"
 	"GuGoTik/src/utils/ptr"
 	"GuGoTik/src/utils/rabbitmq"
 	"context"
 	"encoding/json"
 	"fmt"
 
+	grpc2 "GuGoTik/src/utils/grpc"
 	"time"
 
 	"github.com/go-redis/redis_rate/v10"
@@ -79,14 +79,46 @@ func (c MessageServiceImpl) New() {
 
 	_, err = channel.QueueDeclare(
 		strings.MessageActionEvent,
-		false, false, false, false,
+		true, false, false, false,
 		nil,
 	)
 
 	if err != nil {
 		failOnError(err, "Failed to define queue")
 	}
 
+	_, err = channel.QueueDeclare(
+		strings.MessageGptActionEvent,
+		true, false, false, false,
+		nil,
+	)
+
+	if err != nil {
+		failOnError(err, "Failed to define queue")
+	}
+
+	err = channel.QueueBind(
+		strings.MessageActionEvent,
+		strings.MessageActionEvent,
+		strings.MessageExchange,
+		false,
+		nil,
+	)
+	if err != nil {
+		failOnError(err, "Failed to bind queue to exchange")
+	}
+
+	err = channel.QueueBind(
+		strings.MessageGptActionEvent,
+		strings.MessageGptActionEvent,
+		strings.MessageExchange,
+		false,
+		nil,
+	)
+	if err != nil {
+		failOnError(err, "Failed to bind queue  to exchange")
+	}
+
 	userRpcConn := grpc2.Connect(config.UserRpcServerName)
 
 	userClient = user.NewUserServiceClient(userRpcConn)
@@ -359,13 +391,26 @@ func addMessage(ctx context.Context, fromUserId uint32, toUserId uint32, Context
 		return
 	}
 	headers := rabbitmq.InjectAMQPHeaders(ctx)
-	err = channel.Publish("", strings.MessageActionEvent, false, false,
-		amqp.Publishing{
-			DeliveryMode: amqp.Persistent,
-			ContentType:  "text/plain",
-			Body:         body,
-			Headers:      headers,
-		})
+
+	if message.ToUserId == config.EnvCfg.MagicUserId {
+		err = channel.Publish("", strings.MessageGptActionEvent, false, false,
+			amqp.Publishing{
+				DeliveryMode: amqp.Persistent,
+				ContentType:  "text/plain",
+				Body:         body,
+				Headers:      headers,
+			})
+
+	} else {
+
+		err = channel.Publish("", strings.MessageActionEvent, false, false,
+			amqp.Publishing{
+				DeliveryMode: amqp.Persistent,
+				ContentType:  "text/plain",
+				Body:         body,
+				Headers:      headers,
+			})
+	}
 
 	// result := database.Client.WithContext(ctx).Create(&message)
 

src/services/message/main.go

@@ -11,14 +11,15 @@ import (
 	"GuGoTik/src/utils/logging"
 	"GuGoTik/src/utils/prom"
 	"context"
-	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
-	"github.com/oklog/run"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"net"
 	"net/http"
 	"os"
 	"syscall"
 
+	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
+	"github.com/oklog/run"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
 	"github.com/sirupsen/logrus"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
@@ -74,7 +75,7 @@ func main() {
 	var probe healthImpl.ProbeImpl
 	chat.RegisterChatServiceServer(s, srv)
 	health.RegisterHealthServer(s, &probe)
-
+	defer CloseMQConn()
 	srv.New()
 	srvMetrics.InitializeMetrics(s)