init

.gitignore

@@ -0,0 +1,23 @@
+.DS_Store
+node_modules
+/dist
+
+
+# local env files
+.env.local
+.env.*.local
+
+# Log files
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+
+# Editor directories and files
+.idea
+.vscode
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?

app/api/api.go

@@ -0,0 +1,184 @@
+package api
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/chengjoey/chatadmin/pkg/ssevent"
+
+	"github.com/chengjoey/chatadmin/app/def"
+	"github.com/chengjoey/chatadmin/global"
+	"nhooyr.io/websocket"
+
+	"github.com/chengjoey/chatadmin/controller"
+	"github.com/chengjoey/chatadmin/pkg/response"
+
+	"github.com/gin-gonic/gin"
+)
+
+func AuthRequired() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		token := c.DefaultQuery("token", "")
+		if token == "" {
+			token = c.Request.Header.Get("Authorization")
+		}
+		if token == "" {
+			c.Error(response.WrapError(nil, "token出错"))
+			return
+		}
+		user, err := controller.DescryptTokenToUser(token)
+		if err != nil {
+			c.Error(response.WrapError(err, "身份出错"))
+			return
+		}
+		c.Set("user", user)
+	}
+}
+
+func getUser(c *gin.Context) (*controller.User, error) {
+	data, _ := c.Get("user")
+	user, ok := data.(controller.User)
+	if !ok {
+		return &user, fmt.Errorf("获取用户失败")
+	}
+	return &user, nil
+}
+
+func HomeHandle(c *gin.Context) {
+	c.HTML(http.StatusOK, "index.html", nil)
+}
+
+func Login(c *gin.Context) {
+	var li def.UserLogin
+	if err := c.ShouldBindJSON(&li); err != nil {
+		c.Error(response.WrapError(err, "请求参数格式错误"))
+		return
+	}
+	if l := len(li.Nickname); l < 2 || l > 20 {
+		c.Error(response.WrapError(nil, "非法昵称: %s, 昵称长度:2-20", li.Nickname))
+		return
+	}
+	role := controller.RoleUser
+	isAdmin := controller.IsAdministrator(li.Nickname)
+	if isAdmin {
+		role = controller.RoleAdmin
+	}
+	user := controller.NewUser(li.Nickname, role)
+	c.Header("token", user.Token)
+	user.Token = ""
+	c.JSON(http.StatusOK, gin.H{"value": user, "msg": "登录成功"})
+}
+
+func WebSocketHandle(c *gin.Context) {
+	user, err := getUser(c)
+	if err != nil {
+		c.Error(response.WrapError(nil, "获取用户信息失败"))
+		return
+	}
+	client := c.Query("client")
+	if err := user.CanEnterRoot(client); err != nil {
+		c.Error(response.WrapError(err, "加入聊天失败"))
+		return
+	}
+	options := websocket.AcceptOptions{InsecureSkipVerify: true}
+	conn, err := websocket.Accept(c.Writer, c.Request, &options)
+	if err != nil {
+		c.Error(err)
+		return
+	}
+	defer conn.Close(websocket.StatusInternalError, "内部出错")
+
+	user.Init(conn, client)
+	// token := c.Query("token")
+	// nickname := c.Query("nickname")
+	// to := c.Query("to")
+	// if l := len(nickname); l < 2 || l > 20 {
+	// 	wsjson.Write(c.Request.Context(), conn, response.AppError{Msg: "非法名称，名称长度：2-20"})
+	// 	conn.Close(websocket.StatusUnsupportedData, "nickname illegal!")
+	// 	return
+	// }
+	// if ok, _ := controller.Broadcaster.IsInRoom(nickname); ok {
+	// 	wsjson.Write(c.Request.Context(), conn, response.AppError{Msg: "该名称已经存在"})
+	// 	conn.Close(websocket.StatusUnsupportedData, "nickname exists")
+	// 	return
+	// }
+
+	// user := controller.NewUser(conn, token, nickname, c.Request.RemoteAddr, to)
+	// user.Token = ""
+
+	controller.Broadcaster.UserEntering(user)
+
+	go user.SendMessage(c.Request.Context())
+
+	controller.Broadcaster.SendWelcomeMsg(user, c.Request.Context())
+
+	err = user.RecieveMessage(c.Request.Context())
+
+	controller.Broadcaster.UserLeaving(user)
+
+	if err == nil {
+		conn.Close(websocket.StatusNormalClosure, "")
+	} else {
+		conn.Close(websocket.StatusInternalError, "Read from client error")
+	}
+}
+
+func StreamSSEvent(c *gin.Context) {
+	user, err := getUser(c)
+	if err != nil {
+		c.Error(response.WrapError(nil, "获取用户信息失败"))
+		return
+	}
+	if user.Role != global.RoleAdmin {
+		c.Error(response.WrapError(nil, "权限不足"))
+		return
+	}
+	listener := controller.StreamOpenListener(user.UID)
+	defer controller.StreamCloseListener(user.UID, listener)
+
+	clientGone := c.Writer.CloseNotify()
+	c.Stream(func(w io.Writer) bool {
+		select {
+		case <-clientGone:
+			fmt.Println("用户离开")
+			return false
+		case message := <-listener:
+			sseMsg, ok := message.(*ssevent.Message)
+			if !ok {
+				return false
+			}
+			c.SSEvent("message", sseMsg.Text)
+			return true
+		}
+	})
+}
+
+func AdminGetUserList(c *gin.Context) {
+	user, err := getUser(c)
+	if err != nil {
+		c.Error(response.WrapError(nil, "获取用户信息失败"))
+		return
+	}
+	if user.Role != global.RoleAdmin {
+		c.Error(response.WrapError(nil, "权限不足"))
+		return
+	}
+	users := controller.Broadcaster.GetUserList()
+	c.JSON(http.StatusOK, gin.H{"msg": "获取用户列表成功", "value": users})
+}
+
+func GetHistoryMsg(c *gin.Context) {
+	user, err := getUser(c)
+	if err != nil {
+		c.Error(response.WrapError(nil, "获取用户信息失败"))
+		return
+	}
+	client := c.Query("client")
+	msgs, err := user.GetAllHistoryMsg(client)
+	if err != nil {
+		c.Error(response.WrapError(err, "获取历史消息列表失败"))
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"msg": "获取历史消息列表成功", "value": msgs})
+}

app/def/def.go

@@ -0,0 +1,6 @@
+package def
+
+type UserLogin struct {
+	Nickname string `json:"nickname"`
+	Client   string `json:"client"`
+}

cmd/redis-test/client.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/go-redis/redis/v7"
+)
+
+var rdb *redis.Client
+
+func init() {
+	rdb = redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	// defer rdb.Close()
+}
+
+func main() {
+	// fmt.Println(rdb.HGet("users", "z").Result())
+	fmt.Println(rdb.XRange("user:zc:channel", "-", "+").Result())
+	// status := rdb.XGroupCreate("stm", "stm-group", "$")
+	// fmt.Println(status.Result())
+	// args := &redis.XReadArgs{Streams: []string{"stm", "$"}, Block: time.Duration(0), Count: 2}
+	// for {
+	// 	stm, err := rdb.XRead(args).Result()
+	// 	if err != nil {
+	// 		panic(err)
+	// 	}
+	// 	s := stm[0]
+	// 	for _, msg := range s.Messages {
+	// 		fmt.Println(msg.Values)
+	// 	}
+	// }
+}

cmd/redis-test/server.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+
+	"github.com/go-redis/redis/v7"
+)
+
+var rdb *redis.Client
+
+func init() {
+	rdb = redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	// defer rdb.Close()
+}
+
+func main() {
+	scanner := bufio.NewScanner(os.Stdin)
+	for scanner.Scan() {
+		data := map[string]interface{}{"content": scanner.Text()}
+		arg := &redis.XAddArgs{Stream: "stm", ID: "*", Values: data}
+		fmt.Println(rdb.XAdd(arg).Result())
+	}
+}

cmd/websocket/client.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+
+	"nhooyr.io/websocket/wsjson"
+
+	"nhooyr.io/websocket"
+)
+
+func main() {
+	ctx := context.Background()
+
+	token, to := os.Args[1], os.Args[2]
+	url := fmt.Sprintf("ws://localhost:9090/v1/chat?client=%s", to)
+	options := &websocket.DialOptions{HTTPHeader: http.Header{"Authorization": []string{token}}}
+	c, _, err := websocket.Dial(ctx, url, options)
+	if err != nil {
+		panic(err)
+	}
+	defer c.Close(websocket.StatusInternalError, "内部出错")
+	data := map[string]string{
+		"content": "test send msg",
+	}
+
+	err = wsjson.Write(ctx, c, data)
+	if err != nil {
+		panic(err)
+	}
+	// var v interface{}
+	var v map[string]interface{}
+	go func() {
+		for {
+			err = wsjson.Read(ctx, c, &v)
+			if err != nil {
+				panic(err)
+			}
+			fmt.Printf("Accept from server: %v\n", v)
+		}
+	}()
+	scanner := bufio.NewScanner(os.Stdin)
+	for scanner.Scan() {
+		var data map[string]string
+		_ = json.Unmarshal([]byte(strings.Replace(scanner.Text(), "\n", "", 0)), &data)
+		wsjson.Write(ctx, c, data)
+	}
+	c.Close(websocket.StatusNormalClosure, "")
+}

config/chatadmin.yaml

@@ -0,0 +1,3 @@
+token-secret: 1zrvOpl5u14Y
+
+redis: localhost:6379