feat: analysis user auth and ws

2026-10-15 09:23:07 +08:00
parent 6fa185003d
commit a65b6d2803
102 changed files with 4966 additions and 786 deletions
@@ -0,0 +1,16 @@
+package config
+
+import "github.com/zeromicro/go-zero/rest"
+
+type Config struct {
+	rest.RestConf `json:",optional"`
+	Nacos         NacosConf
+}
+
+type NacosConf struct {
+	Hosts       []string
+	NamespaceId string `json:",optional"`
+	Group       string `json:",optional"`
+	RegisterIP  string `json:",optional"`
+	ConfigID    string `json:",optional"`
+}
@@ -0,0 +1,26 @@
+package handler
+
+import (
+	"net/http"
+
+	"lone-services/services/ws/internal/logic"
+	"lone-services/services/ws/internal/svc"
+	"lone-services/services/ws/internal/types"
+
+	"github.com/zeromicro/go-zero/rest/httpx"
+)
+
+func ConnectHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var req types.ConnectReq
+		if err := httpx.Parse(r, &req); err != nil {
+			httpx.ErrorCtx(r.Context(), w, err)
+			return
+		}
+
+		l := logic.NewConnectLogic(r.Context(), svcCtx)
+		if err := l.Connect(w, r, &req); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+		}
+	}
+}
@@ -0,0 +1,24 @@
+// Code generated by goctl. DO NOT EDIT.
+// goctl 1.10.1
+
+package handler
+
+import (
+	"net/http"
+
+	"lone-services/services/ws/internal/svc"
+
+	"github.com/zeromicro/go-zero/rest"
+)
+
+func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
+	server.AddRoutes(
+		[]rest.Route{
+			{
+				Method:  http.MethodGet,
+				Path:    "/ws",
+				Handler: ConnectHandler(serverCtx),
+			},
+		},
+	)
+}
@@ -0,0 +1,126 @@
+package logic
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"lone-services/pkg/redis"
+	pkgws "lone-services/pkg/ws"
+	"lone-services/services/ws/internal/svc"
+	"lone-services/services/ws/internal/types"
+
+	"github.com/gorilla/websocket"
+	jsoniter "github.com/json-iterator/go"
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
+
+type ConnectLogic struct {
+	logx.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+func NewConnectLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ConnectLogic {
+	return &ConnectLogic{
+		Logger: logx.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *ConnectLogic) Connect(w http.ResponseWriter, r *http.Request, req *types.ConnectReq) error {
+	if req.SessionId == "" {
+		return fmt.Errorf("missing session_id")
+	}
+
+	sess, err := pkgws.GetSession(l.ctx, req.SessionId)
+	if err != nil {
+		if errors.Is(err, redis.Nil) {
+			return fmt.Errorf("qrcode session expired")
+		}
+		l.Errorf("get session: %v", err)
+		return fmt.Errorf("qrcode session invalid")
+	}
+
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		return err
+	}
+
+	client := &svc.Client{SessionID: req.SessionId, Conn: conn}
+	l.svcCtx.Hub.Add(client)
+
+	if err := l.pushSnapshot(client, sess); err != nil {
+		l.Errorf("push snapshot session=%s: %v", req.SessionId, err)
+		l.svcCtx.Hub.Remove(req.SessionId, conn)
+		_ = conn.Close()
+		return nil
+	}
+
+	go l.readLoop(client)
+	go l.autoClose(client)
+	return nil
+}
+
+func (l *ConnectLogic) pushSnapshot(c *svc.Client, sess *pkgws.Session) error {
+	var data interface{}
+	switch sess.Status {
+	case pkgws.StatusPending:
+		data = nil
+	case pkgws.StatusScanned, pkgws.StatusLogined:
+		if len(sess.Data) > 0 && string(sess.Data) != "null" {
+			var parsed any
+			if err := jsoniter.Unmarshal(sess.Data, &parsed); err != nil {
+				data = jsoniter.RawMessage(sess.Data)
+			} else {
+				data = parsed
+			}
+		} else if sess.Status == pkgws.StatusScanned {
+			data = map[string]int{"ws_type": pkgws.WsTypeScanned}
+		} else {
+			data = nil
+		}
+	default:
+		data = nil
+	}
+
+	payload, err := jsoniter.Marshal(pkgws.Success(data))
+	if err != nil {
+		return err
+	}
+	l.svcCtx.Hub.Send(c.SessionID, payload)
+	return nil
+}
+
+func (l *ConnectLogic) readLoop(c *svc.Client) {
+	defer func() {
+		_ = c.Conn.Close()
+		l.svcCtx.Hub.Remove(c.SessionID, c.Conn)
+	}()
+
+	for {
+		if _, _, err := c.Conn.ReadMessage(); err != nil {
+			return
+		}
+	}
+}
+
+func (l *ConnectLogic) autoClose(c *svc.Client) {
+	timer := time.NewTimer(pkgws.SessionTTL)
+	defer timer.Stop()
+	<-timer.C
+	l.Infof("timeout close session=%s", c.SessionID)
+	_ = c.Conn.Close()
+	l.svcCtx.Hub.Remove(c.SessionID, c.Conn)
+}
@@ -0,0 +1,62 @@
+package svc
+
+import (
+	"sync"
+
+	"github.com/gorilla/websocket"
+)
+
+type Client struct {
+	SessionID string
+	Conn      *websocket.Conn
+	sendMu    sync.Mutex
+}
+
+type Hub struct {
+	mu      sync.RWMutex
+	clients map[string]*Client
+}
+
+func NewHub() *Hub {
+	return &Hub{clients: make(map[string]*Client)}
+}
+
+func (h *Hub) Add(c *Client) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if old, ok := h.clients[c.SessionID]; ok && old.Conn != c.Conn {
+		_ = old.Conn.Close()
+	}
+	h.clients[c.SessionID] = c
+}
+
+func (h *Hub) Remove(sessionID string, conn *websocket.Conn) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	cur, ok := h.clients[sessionID]
+	if !ok {
+		return
+	}
+	if conn != nil && cur.Conn != conn {
+		return
+	}
+	delete(h.clients, sessionID)
+}
+
+func (h *Hub) Send(sessionID string, payload []byte) bool {
+	h.mu.RLock()
+	c, ok := h.clients[sessionID]
+	h.mu.RUnlock()
+	if !ok {
+		return false
+	}
+
+	c.sendMu.Lock()
+	defer c.sendMu.Unlock()
+	if err := c.Conn.WriteMessage(websocket.TextMessage, payload); err != nil {
+		_ = c.Conn.Close()
+		h.Remove(sessionID, c.Conn)
+		return false
+	}
+	return true
+}
@@ -0,0 +1,31 @@
+package svc
+
+import (
+	"context"
+
+	"lone-services/services/ws/internal/config"
+)
+
+type ServiceContext struct {
+	Config config.Config
+	Hub    *Hub
+	cancel context.CancelFunc
+}
+
+func NewServiceContext(c config.Config) *ServiceContext {
+	h := NewHub()
+	ctx, cancel := context.WithCancel(context.Background())
+	go NewSubscriber(h).Run(ctx)
+
+	return &ServiceContext{
+		Config: c,
+		Hub:    h,
+		cancel: cancel,
+	}
+}
+
+func (s *ServiceContext) Close() {
+	if s.cancel != nil {
+		s.cancel()
+	}
+}
@@ -0,0 +1,78 @@
+package svc
+
+import (
+	"context"
+
+	"lone-services/pkg/redis"
+	pkgws "lone-services/pkg/ws"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type Subscriber struct {
+	hub *Hub
+}
+
+func NewSubscriber(h *Hub) *Subscriber {
+	return &Subscriber{hub: h}
+}
+
+func (s *Subscriber) Run(ctx context.Context) {
+	if redis.Client == nil {
+		logx.Error("ws pubsub: redis not initialized")
+		return
+	}
+
+	sub := redis.Client.Subscribe(ctx, pkgws.RedisChannelQRPush)
+	defer sub.Close()
+
+	ch := sub.Channel()
+	logx.Infof("ws pubsub: subscribed %s", pkgws.RedisChannelQRPush)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg, ok := <-ch:
+			if !ok {
+				return
+			}
+			s.handle([]byte(msg.Payload))
+		}
+	}
+}
+
+func (s *Subscriber) handle(raw []byte) {
+	var evt pkgws.Event
+	if err := jsoniter.Unmarshal(raw, &evt); err != nil {
+		logx.Errorf("ws pubsub: unmarshal event: %v", err)
+		return
+	}
+	if evt.SessionID == "" {
+		return
+	}
+
+	resp := pkgws.Response{
+		Code:    evt.Code,
+		Message: evt.Message,
+		Data:    nil,
+	}
+	if len(evt.Data) > 0 && string(evt.Data) != "null" {
+		var data any
+		if err := jsoniter.Unmarshal(evt.Data, &data); err != nil {
+			resp.Data = jsoniter.RawMessage(evt.Data)
+		} else {
+			resp.Data = data
+		}
+	}
+
+	payload, err := jsoniter.Marshal(resp)
+	if err != nil {
+		logx.Errorf("ws pubsub: marshal response: %v", err)
+		return
+	}
+	if !s.hub.Send(evt.SessionID, payload) {
+		logx.Infof("ws pubsub: no local conn for session=%s", evt.SessionID)
+	}
+}
@@ -0,0 +1,8 @@
+// Code generated by goctl. DO NOT EDIT.
+// goctl 1.10.1
+
+package types
+
+type ConnectReq struct {
+	SessionId string `form:"session_id"`
+}