temp

apps/mqs/etc/mqs.dev.yaml

@@ -22,4 +22,10 @@ KqConsumerCharge:
   Topic: apirequest
   Offset: first
   Consumers: 2  # 同样分配2个消费者
-  Processors: 2
+  Processors: 2
+
+UserRpc:
+  Etcd:
+    Hosts:
+      - 127.0.0.1:2379
+    Key: user.rpc

apps/mqs/etc/mqs.yaml

@@ -22,4 +22,10 @@ KqConsumerCharge:
   Topic: apirequest
   Offset: first
   Consumers: 2  # 同样分配2个消费者
-  Processors: 2
+  Processors: 2
+
+UserRpc:
+  Etcd:
+    Hosts:
+      - tyapi_etcd:2379
+    Key: user.rpc

apps/mqs/internal/config/config.go

@@ -3,10 +3,12 @@ package config
 import (
 	"github.com/zeromicro/go-queue/kq"
 	"github.com/zeromicro/go-zero/rest"
+	"github.com/zeromicro/go-zero/zrpc"
 )
 
 type Config struct {
 	rest.RestConf
 	KqConsumerLog    kq.KqConf
 	KqConsumerCharge kq.KqConf
+	UserRpc          zrpc.RpcClientConf
 }

apps/mqs/internal/mqs/apirequest/charge.go

@@ -2,8 +2,11 @@ package apirequest
 
 import (
 	"context"
+	"encoding/json"
 	"github.com/zeromicro/go-zero/core/logx"
 	"tianyuan-api/apps/mqs/internal/svc"
+	"tianyuan-api/apps/user/user"
+	"tianyuan-api/pkg/models"
 )
 
 type Charge struct {
@@ -19,6 +22,18 @@ func NewCharge(ctx context.Context, svcCtx *svc.ServiceContext) *Charge {
 }
 
 func (l *Charge) Consume(ctx context.Context, key, val string) error {
-	logx.Infof("Charge key :%s , val :%s", key, val)
+	var apiRequestMessage models.ApiRequestMessage
+	err := json.Unmarshal([]byte(val), &apiRequestMessage)
+	if err != nil {
+		return err
+	}
+	logx.Infof("Charge val :%v", apiRequestMessage)
+	if !apiRequestMessage.Charges {
+		return nil
+	}
+	_, updateWalletErr := l.svcCtx.WalletRpc.UpdateWallet(ctx, &user.UpdateWalletRequest{UserId: apiRequestMessage.UserId, TransactionId: apiRequestMessage.TransactionID, ProductCode: apiRequestMessage.ProductCode, Remark: apiRequestMessage.Remark})
+	if updateWalletErr != nil {
+		return err
+	}
 	return nil
 }

apps/mqs/internal/mqs/apirequest/log.go

@@ -2,8 +2,11 @@ package apirequest
 
 import (
 	"context"
-	"github.com/zeromicro/go-zero/core/logx"
+	"encoding/json"
 	"tianyuan-api/apps/mqs/internal/svc"
+	"tianyuan-api/apps/user/user"
+	"tianyuan-api/pkg/models"
+	"time"
 )
 
 type Log struct {
@@ -19,6 +22,22 @@ func NewLog(ctx context.Context, svcCtx *svc.ServiceContext) *Log {
 }
 
 func (l *Log) Consume(ctx context.Context, key, val string) error {
-	logx.Infof("log key :%s , val :%s", key, val)
+	var apiRequestMessage models.ApiRequestMessage
+	err := json.Unmarshal([]byte(val), &apiRequestMessage)
+	if err != nil {
+		return err
+	}
+	_, err = l.svcCtx.ApiRequestRpc.AddApiRequest(ctx, &user.AddApiRequestRequest{
+		TransactionId: apiRequestMessage.TransactionID,
+		UserId:        apiRequestMessage.UserId,
+		ProductCode:   apiRequestMessage.ProductCode,
+		Status:        apiRequestMessage.Status,
+		Charges:       apiRequestMessage.Charges,
+		Remark:        apiRequestMessage.Remark,
+		Timestamp:     apiRequestMessage.Timestamp.Format(time.RFC3339Nano),
+	})
+	if err != nil {
+		return err
+	}
 	return nil
 }

apps/mqs/internal/svc/servicecontext.go

@@ -1,13 +1,24 @@
 package svc
 
-import "tianyuan-api/apps/mqs/internal/config"
+import (
+	"github.com/zeromicro/go-zero/zrpc"
+	"tianyuan-api/apps/mqs/internal/config"
+	"tianyuan-api/apps/user/user"
+)
 
 type ServiceContext struct {
-	Config config.Config
+	Config        config.Config
+	WalletRpc     user.WalletServiceClient
+	ApiRequestRpc user.ApiRequestServiceClient
 }
 
 func NewServiceContext(c config.Config) *ServiceContext {
+	walletRpc := user.NewWalletServiceClient(zrpc.MustNewClient(c.UserRpc).Conn())
+	apiRequestRpc := user.NewApiRequestServiceClient(zrpc.MustNewClient(c.UserRpc).Conn())
+
 	return &ServiceContext{
-		Config: c,
+		Config:        c,
+		WalletRpc:     walletRpc,
+		ApiRequestRpc: apiRequestRpc,
 	}
 }