474 lines
13 KiB
Go
474 lines
13 KiB
Go
package service
|
||
|
||
import (
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"github.com/golang-jwt/jwt"
|
||
"log"
|
||
"regexp"
|
||
"strconv"
|
||
"time"
|
||
"videoplayer/dao"
|
||
"videoplayer/proto"
|
||
"videoplayer/worker"
|
||
)
|
||
|
||
func CreateUser(name, password, email, gender string, age int) uint {
|
||
id := dao.CreateUser(name, password, email, gender, age)
|
||
if id != 0 {
|
||
//添加用户信息到同步列表
|
||
err := setSyncUserDataSet("add", int(id))
|
||
if err != nil {
|
||
return id
|
||
}
|
||
}
|
||
return id
|
||
}
|
||
|
||
func GetUser(name, email, password string) dao.User {
|
||
emailRegex := `^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$`
|
||
re := regexp.MustCompile(emailRegex)
|
||
var user dao.User
|
||
if re.MatchString(name) {
|
||
user = dao.FindUserByEmail(name)
|
||
} else {
|
||
user = dao.FindUserByName(name)
|
||
}
|
||
if user.ID != 0 && user.Password == password {
|
||
return user
|
||
}
|
||
return dao.User{}
|
||
}
|
||
|
||
func ContainsUser(name, email string) bool {
|
||
user := dao.FindUserByName(name)
|
||
user2 := dao.FindUserByEmail(email)
|
||
if user.ID != 0 || user2.ID != 0 {
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
func GetUserByID(id int) []proto.User {
|
||
return dao.FindUserByID(id)
|
||
}
|
||
|
||
// 获取用户信息，有redis缓存
|
||
func GetUserByIDWithCache(id int) dao.User {
|
||
if id <= 0 {
|
||
return dao.User{}
|
||
}
|
||
var user dao.User
|
||
//先从redis获取
|
||
key := "user_info_" + strconv.Itoa(id)
|
||
user_str := worker.GetRedis(key)
|
||
if user_str != "" {
|
||
err := json.Unmarshal([]byte(user_str), &user)
|
||
if err != nil {
|
||
fmt.Println("get user info , json unmarshal error:", err, "\tuser_str:", user_str)
|
||
return dao.User{}
|
||
}
|
||
} else {
|
||
user = dao.FindUserByID2(id)
|
||
if user.ID != 0 {
|
||
userJson, err := json.Marshal(user)
|
||
if err != nil {
|
||
fmt.Println("get user info , json marshal error:", err)
|
||
return dao.User{}
|
||
}
|
||
success := worker.SetRedis(key, string(userJson))
|
||
if !success {
|
||
fmt.Println("set redis error,user json:", string(userJson))
|
||
}
|
||
}
|
||
}
|
||
return user
|
||
}
|
||
|
||
func GetUserByNameLike(name string) []proto.User {
|
||
return dao.FindUserByNameLike(name)
|
||
}
|
||
|
||
func UpdateUser(user_id int, req proto.UpdateUserInfoReq) (int, error) {
|
||
cur_user := dao.FindUserByID2(user_id)
|
||
//fmt.Println("cur_user:", cur_user, "req:", req)
|
||
if user_id == req.ID && cur_user.Role != "admin" {
|
||
err := dao.UpdateUserByID3(user_id, req) //用户修改自己的信息，不能修改权限信息
|
||
//添加修改用户信息到同步列表
|
||
if err == nil {
|
||
err2 := setSyncUserDataSet("update", user_id)
|
||
UpdateUserCache(user_id)
|
||
if err2 != nil {
|
||
fmt.Println("set sync user data set error:", err2)
|
||
return user_id, nil
|
||
}
|
||
}
|
||
return user_id, err
|
||
} else if cur_user.Role == "admin" {
|
||
err := dao.UpdateUserByID2(req.ID, req)
|
||
if err == nil {
|
||
//添加修改用户信息到同步列表
|
||
err2 := setSyncUserDataSet("update", req.ID)
|
||
UpdateUserCache(req.ID)
|
||
if err2 != nil {
|
||
fmt.Println("set sync user data set error:", err2)
|
||
return req.ID, nil
|
||
}
|
||
}
|
||
return req.ID, nil
|
||
} else {
|
||
return 0, nil
|
||
}
|
||
}
|
||
|
||
func UpdateUserCache(id int) {
|
||
key := "user_info_" + strconv.Itoa(id)
|
||
if worker.IsContainKey(key) {
|
||
user := GetUserByID(id)
|
||
userJson, err := json.Marshal(user)
|
||
if err != nil {
|
||
fmt.Println("get user info , json marshal error:", err)
|
||
}
|
||
success := worker.SetRedis(key, string(userJson))
|
||
if !success {
|
||
fmt.Println("set redis error,user json:", string(userJson))
|
||
}
|
||
}
|
||
}
|
||
|
||
func DeleteUserService(id, user_id int) int {
|
||
res := 0
|
||
if user_id == id {
|
||
res = dao.DeleteUserByID(id)
|
||
} else {
|
||
user := dao.FindUserByID2(user_id)
|
||
if user.Role == "admin" {
|
||
res = dao.DeleteUserByID(id)
|
||
}
|
||
}
|
||
if res != 0 {
|
||
//添加删除用户信息到同步列表
|
||
err := setSyncUserDataSet("delete", id)
|
||
if err != nil {
|
||
return res
|
||
}
|
||
}
|
||
return res
|
||
}
|
||
func UserSyncDataFromMaster() {
|
||
//从接口获取数据
|
||
url := "https://" + proto.Config.MASTER_SERVER_DOMAIN + "/user/sync"
|
||
tokens := worker.GetRedisSetMembers("super_permission_tokens")
|
||
var req proto.SyncUserReq
|
||
req.Token = tokens[0]
|
||
req.Device = proto.Config.SERVER_NAME
|
||
all := worker.GetRedis("user_sync_all")
|
||
is_all := false //是否全量同步
|
||
if all == "" || all == "1" {
|
||
is_all = true
|
||
//清空数据表
|
||
err := dao.ClearAllUsers()
|
||
if err != nil {
|
||
fmt.Println("All ClearAllUsers error:", err)
|
||
return
|
||
}
|
||
worker.SetRedisForever("user_sync_all", "1")
|
||
req.Types = 1
|
||
} else {
|
||
worker.SetRedisForever("user_sync_all", "2")
|
||
req.Types = 2
|
||
}
|
||
|
||
user_sync_data, err := worker.SyncDataFromMasterReq2(url, req)
|
||
if err != nil {
|
||
fmt.Println("UserSyncDataFromMaster error:", err)
|
||
return
|
||
}
|
||
add_users := user_sync_data.Add
|
||
update_users := user_sync_data.Update
|
||
delete_users := user_sync_data.Delete
|
||
//未成功操作的id
|
||
var fail_ids []uint
|
||
|
||
//添加用户
|
||
var add_confirm []proto.UserConfirmID
|
||
for _, v := range add_users {
|
||
res := dao.AddUserSync(v)
|
||
if res == 0 {
|
||
fail_ids = append(fail_ids, v.ID)
|
||
} else {
|
||
add_confirm = append(add_confirm, proto.UserConfirmID{ID: v.ID})
|
||
}
|
||
}
|
||
//更新用户
|
||
var update_confirm []proto.UserConfirmID
|
||
for _, v := range update_users {
|
||
res := dao.UpdateUserSync(v)
|
||
if res == 0 {
|
||
fail_ids = append(fail_ids, v.ID)
|
||
} else {
|
||
update_confirm = append(update_confirm, proto.UserConfirmID{ID: v.ID})
|
||
}
|
||
}
|
||
//删除用户
|
||
var delete_confirm []proto.UserConfirmID
|
||
for _, v := range delete_users {
|
||
res := dao.DeleteUserSync(v)
|
||
if res == 0 {
|
||
fail_ids = append(fail_ids, v.ID)
|
||
} else {
|
||
delete_confirm = append(delete_confirm, proto.UserConfirmID{ID: v.ID})
|
||
}
|
||
}
|
||
|
||
//确认同步数据
|
||
if is_all == false {
|
||
var data proto.UserSyncConfirm
|
||
data.Add = add_confirm
|
||
data.Update = update_confirm
|
||
data.Delete = delete_confirm
|
||
//确认同步数据请求
|
||
var confirm_req proto.SyncUserReq
|
||
confirm_req.Token = tokens[0]
|
||
confirm_req.Device = proto.Config.SERVER_NAME
|
||
confirm_req.Types = 3
|
||
confirm_req.Confirm = data
|
||
worker.SyncDataFromMasterReq2(url, confirm_req)
|
||
} else {
|
||
worker.SetRedisForever("user_sync_all", "2")
|
||
}
|
||
}
|
||
|
||
// 同步数据到主服务器-增删改数据
|
||
func GetUserSyncData(device string) dao.UserSyncResp {
|
||
key := device + "_sync_user_ids"
|
||
add_temp_key := device + "_sync_user_ids_add_confirm_temp"
|
||
update_temp_key := device + "_sync_user_ids_update_confirm_temp"
|
||
delete_temp_key := device + "_sync_user_ids_delete_confirm_temp"
|
||
//需要获取暂存集合的并集，清空暂存集合，存入待确认集合
|
||
add_user_ids := worker.GetRedisSetUnion(key+"_add", add_temp_key)
|
||
update_user_ids := worker.GetRedisSetUnion(key+"_update", update_temp_key)
|
||
delete_user_ids := worker.GetRedisSetUnion(key+"_delete", delete_temp_key)
|
||
add_users := []dao.User{}
|
||
update_users := []dao.User{}
|
||
delete_users := []proto.UserDelID{}
|
||
for _, v := range add_user_ids {
|
||
id, _ := strconv.Atoi(v)
|
||
user := dao.FindUserByUserID(id)
|
||
add_users = append(add_users, user)
|
||
}
|
||
|
||
for _, v := range update_user_ids {
|
||
id, _ := strconv.Atoi(v)
|
||
user := dao.FindUserByUserID(id)
|
||
update_users = append(update_users, user)
|
||
}
|
||
|
||
for _, v := range delete_user_ids {
|
||
id, _ := strconv.Atoi(v)
|
||
delete_users = append(delete_users, proto.UserDelID{ID: uint(id)})
|
||
}
|
||
//将id存入暂存集合，清空原集合，存入待确认集合主要保证在确认时，有新的数据加入不会在确认时漏掉
|
||
worker.SetRedisSetUnionAndStore(add_temp_key, key+"_add")
|
||
worker.ClearRedisSet(key + "_add")
|
||
worker.SetRedisSetUnionAndStore(update_temp_key, key+"_update")
|
||
worker.ClearRedisSet(key + "_update")
|
||
worker.SetRedisSetUnionAndStore(delete_temp_key, key+"_delete")
|
||
worker.ClearRedisSet(key + "_delete")
|
||
return dao.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users}
|
||
}
|
||
|
||
func setSyncUserDataSet(t string, id int) error {
|
||
devices := worker.GetRedisSetMembers("sync_devices_ids") //主服务器查看从服务器的设备列表
|
||
fmt.Println("set sync user data set devices:", devices, "t:", t, "id:", id)
|
||
var err error
|
||
for _, device := range devices {
|
||
key := device + "_sync_user_ids"
|
||
if t == "add" {
|
||
key_ := key + "_add"
|
||
worker.SetRedisSetAdd(key_, strconv.Itoa(id))
|
||
} else if t == "update" {
|
||
key_ := key + "_update"
|
||
worker.SetRedisSetAdd(key_, strconv.Itoa(id))
|
||
} else if t == "delete" {
|
||
key_ := key + "_delete"
|
||
worker.SetRedisSetAdd(key_, strconv.Itoa(id))
|
||
} else {
|
||
err = errors.New("error")
|
||
}
|
||
}
|
||
return err
|
||
}
|
||
|
||
// 确认同步数据
|
||
func ConfirmSyncUserData(device string, data proto.UserSyncConfirm) error {
|
||
var err error
|
||
if len(data.Add) > 0 {
|
||
var ids_add []string
|
||
|
||
for _, v := range data.Add {
|
||
ids_add = append(ids_add, strconv.Itoa(int(v.ID)))
|
||
}
|
||
add_key := device + "_sync_user_ids_add_confirm"
|
||
isSuccess := worker.SetRedisSetAddBatchWithExpire(add_key, ids_add, time.Second*30)
|
||
if !isSuccess {
|
||
err = errors.New("set add confirm error")
|
||
return err
|
||
}
|
||
ids_add_confirm_temp := device + "_sync_user_ids_add_confirm_temp"
|
||
//取差集
|
||
add_diff := worker.SetRedisSetDiffAndStore(ids_add_confirm_temp, add_key)
|
||
if add_diff == false {
|
||
err = errors.New("add diff error")
|
||
return err
|
||
}
|
||
}
|
||
|
||
if len(data.Update) > 0 {
|
||
|
||
var ids_update []string
|
||
for _, v := range data.Update {
|
||
ids_update = append(ids_update, strconv.Itoa(int(v.ID)))
|
||
}
|
||
update_key := device + "_sync_user_ids_update_confirm"
|
||
isSuccess := worker.SetRedisSetAddBatchWithExpire(update_key, ids_update, time.Second*30)
|
||
if !isSuccess {
|
||
err = errors.New("set update confirm error")
|
||
return err
|
||
}
|
||
ids_update_confirm_temp := device + "_sync_user_ids_update_confirm_temp"
|
||
|
||
update_diff := worker.SetRedisSetDiffAndStore(ids_update_confirm_temp, update_key)
|
||
if update_diff == false {
|
||
err = errors.New("update diff error")
|
||
return err
|
||
}
|
||
|
||
}
|
||
|
||
if len(data.Delete) > 0 {
|
||
var ids_delete []string
|
||
for _, v := range data.Delete {
|
||
ids_delete = append(ids_delete, strconv.Itoa(int(v.ID)))
|
||
}
|
||
del_key := device + "_sync_user_ids_delete_confirm"
|
||
isSuccess := worker.SetRedisSetAddBatchWithExpire(del_key, ids_delete, time.Second*30)
|
||
if !isSuccess {
|
||
err = errors.New("set del confirm error")
|
||
return err
|
||
}
|
||
|
||
//待确认集合暂存
|
||
ids_delete_confirm_temp := device + "_sync_user_ids_delete_confirm_temp"
|
||
delete_diff := worker.SetRedisSetDiffAndStore(ids_delete_confirm_temp, del_key)
|
||
if delete_diff == false {
|
||
err = errors.New("delete diff error")
|
||
return err
|
||
}
|
||
}
|
||
return err
|
||
}
|
||
|
||
// 生成新的token，存入redis，返回信息
|
||
func CreateTokenAndSave(user dao.User) (string, error) {
|
||
var tokenString string
|
||
var err error
|
||
key := "user_" + user.Name
|
||
redis_token := worker.GetRedis(string(key))
|
||
if redis_token == "" {
|
||
// 生成 JWT 令牌
|
||
token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
|
||
"username": user.Name,
|
||
"id": user.ID,
|
||
"exp": time.Now().Add(time.Hour * 10).Unix(), // 令牌过期时间, 10小时后过期
|
||
})
|
||
tokenString, err = token.SignedString(proto.SigningKey)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
|
||
worker.SetRedisWithExpire("user_"+user.Name, tokenString, time.Hour*10) // 将用户信息存入
|
||
worker.SetRedisWithExpire(tokenString, tokenString, time.Hour*10) // 设置过期时间为10分钟
|
||
data := make(map[string]interface{})
|
||
data["id"] = user.ID
|
||
data["username"] = user.Name
|
||
data["email"] = user.Email
|
||
worker.SetHash(tokenString, data) // 将用户信息存入
|
||
} else {
|
||
tokenString = redis_token
|
||
}
|
||
// 返回令牌
|
||
return tokenString, err
|
||
}
|
||
|
||
// 获取用户信息，有redis缓存
|
||
func GetUserByIDFromUserCenter(id int) dao.User {
|
||
if id <= 0 {
|
||
return dao.User{}
|
||
}
|
||
var user dao.User
|
||
//先从redis获取
|
||
key := "user_info_" + strconv.Itoa(id)
|
||
userStr := worker.GetRedis(key)
|
||
log.Println("user_str:", userStr)
|
||
if userStr != "" {
|
||
err := json.Unmarshal([]byte(userStr), &user)
|
||
if err != nil {
|
||
fmt.Println("get user info , json unmarshal error:", err, "\tuser_str:", userStr)
|
||
return dao.User{}
|
||
}
|
||
} else {
|
||
user = GetUserInfoByIDFromUserCenterHttp(id)
|
||
if user.ID != 0 {
|
||
userJson, err := json.Marshal(user)
|
||
if err != nil {
|
||
fmt.Println("get user info , json marshal error:", err)
|
||
return dao.User{}
|
||
}
|
||
userStr = string(userJson)
|
||
success := worker.SetRedisWithExpire(key, userStr, time.Second*10)
|
||
if !success {
|
||
fmt.Println("set redis error,user json:", userStr)
|
||
}
|
||
} else {
|
||
log.Println("GetUserByIDFromUserCenter user not found, id:", id)
|
||
log.Println("response user:", user)
|
||
}
|
||
}
|
||
return user
|
||
}
|
||
|
||
type UserInfoResponse struct {
|
||
Code int `json:"code"`
|
||
Message string `json:"message"`
|
||
Data dao.User `json:"data"`
|
||
}
|
||
|
||
func GetUserInfoByIDFromUserCenterHttp(id int) dao.User {
|
||
var resp UserInfoResponse
|
||
url := "https://uc.ljsea.top/user/info?super_id=1"
|
||
tokens := worker.GetRedisSetMembers("super_permission_tokens")
|
||
if len(tokens) == 0 {
|
||
return resp.Data
|
||
}
|
||
token := tokens[0]
|
||
//请求参数
|
||
req := map[string]int{
|
||
"id": id,
|
||
}
|
||
headers := map[string]string{
|
||
"token": token,
|
||
}
|
||
reqByte, _ := json.Marshal(req)
|
||
err, respBytes := worker.DoPostRequestJSON(url, reqByte, headers)
|
||
if err != nil {
|
||
log.Println("GetUserInfoByIDFromUserCenterHttp error:", err)
|
||
return resp.Data
|
||
}
|
||
if err2 := json.Unmarshal(respBytes, &resp); err2 != nil {
|
||
log.Println("GetUserInfoByIDFromUserCenterHttp json unmarshal error:", err2)
|
||
}
|
||
return resp.Data
|
||
}
|