aozhiwei b9a8208c1e 1
2024-07-06 16:32:29 +08:00

627 lines
15 KiB
Go

package mail
import (
"q5"
"f5"
"jccommon"
"main/common"
"mt"
"main/constant"
"sync"
"fmt"
"github.com/gin-gonic/gin"
)
type dbEvent struct {
idx int64
eventName string
param1 string
entry q5.ListHead
}
type userGroup struct {
groupId int64
userHash *sync.Map
}
type mailMgr struct {
idHash sync.Map //int64 => *mail
wholeMails sync.Map //int64 => *mail
groupMails sync.Map //int64 => *mail
personalMails sync.Map //string => sync.Map<int64, *mail>
groupHash sync.Map //int64 => *userGroup
lastSyncEventIdx int64
pullingEvent bool
procingEvent bool
eventQueue q5.Queue
timerWp *f5.TimerWp
}
func (this *mailMgr) Init() {
this.eventQueue.Init()
this.syncEvent()
this.loadMails()
this.loadGroups()
this.loadGroupMembers()
f5.GetApp().RegisterCaHandle("Mail", "getMailList", this.CaGetMailList)
f5.GetApp().RegisterCaHandle("Mail", "markMail", this.CaMarkMail)
f5.GetApp().RegisterCaHandle("Mail", "getUnreadMailCnt", this.CaGetUnreadMailCnt)
f5.GetApp().RegisterCaHandle("Mail", "getAttachment", this.CaGetAttachment)
f5.GetApp().RegisterCaHandle("Mail", "deleteMails", this.CaDeleteMails)
f5.GetApp().RegisterCaHandle("MailMgr", "sendMail", this.CaSendMail)
this.timerWp = f5.GetTimer().SetIntervalWp(
300,
func (e int32, args* q5.Args) {
if e == q5.TIMER_EXEC_EVENT {
this.updateDbEvent()
}
})
}
func (this *mailMgr) UnInit() {
f5.GetTimer().Delete(this.timerWp)
this.timerWp = nil
}
func (this *mailMgr) loadMails() {
f5.GetSysLog().Info("mailMgr.loadMails begin")
nowTime := f5.GetApp().GetNowSeconds()
lastIdx := f5.GetGoStyleDb().SyncBatchLoadFullTable(
constant.MAIL_DB,
"SELECT * FROM t_mail WHERE idx > %d AND deleted = 0 AND expiretime > " + q5.ToString(nowTime),
func (ds *f5.DataSet) {
p := newMail()
p.loadFromDb(ds)
this.addMail(p)
},
func (err error) {
panic(fmt.Sprintf("mailMgr.loadMails dberror:%s", err))
})
f5.GetSysLog().Info("mailMgr.loadMails end lastIdx:%d mailNum:%d",
lastIdx,
q5.GetSyncMapSize(this.idHash))
}
func (this *mailMgr) loadGroups() {
f5.GetSysLog().Info("mailMgr.loadGroups begin")
lastIdx := f5.GetGoStyleDb().SyncBatchLoadFullTable(
constant.MAIL_DB,
"SELECT * FROM t_group WHERE idx > %d AND deleted = 0",
func (ds *f5.DataSet) {
groupId := q5.ToInt64(ds.GetByName("group_id"))
if this.getGroup(groupId) != nil {
panic(fmt.Sprintf("mailMgr.loadGroups group_id error"))
}
p := newUserGroup()
p.groupId = groupId
this.addGroup(p)
},
func (err error) {
panic(fmt.Sprintf("mailMgr.loadGroups dberror:%s", err))
})
f5.GetSysLog().Info("mailMgr.loadGroups end lastIdx:%d mailNum:%d",
lastIdx,
q5.GetSyncMapSize(this.idHash))
}
func (this *mailMgr) loadGroupMembers() {
f5.GetSysLog().Info("mailMgr.loadGroupMembers begin")
lastIdx := f5.GetGoStyleDb().SyncBatchLoadFullTable(
constant.MAIL_DB,
"SELECT * FROM t_member WHERE idx > %d AND deleted = 0",
func (ds *f5.DataSet) {
groupId := q5.ToInt64(ds.GetByName("group_id"))
memberId := ds.GetByName("member_id")
p := this.getGroup(groupId)
if p != nil {
p.userHash.Store(memberId, q5.ToInt64(ds.GetByName("idx")))
}
},
func (err error) {
panic(fmt.Sprintf("mailMgr.loadGroupMembers dberror:%s", err))
})
f5.GetSysLog().Info("mailMgr.loadGroupMembers end lastIdx:%d mailNum:%d",
lastIdx,
q5.GetSyncMapSize(this.idHash))
}
func (this *mailMgr) CaGetMailList(c *gin.Context) {
hum := c.MustGet("hum").(common.Player)
rspObj := struct {
ErrCode int32 `json:"errcode"`
ErrMsg string `json:"errmsg"`
MailList []*common.MailDto `json:"maillist"`
}{
MailList: []*common.MailDto{},
}
this.traversePlayerMail(
hum,
func (m *mail) bool {
if m.IsValid(hum) {
if hum.IsReadable(m) {
mailDto := new(common.MailDto)
if m.fillMailDto(hum, mailDto) {
q5.AppendSlice(&rspObj.MailList, mailDto)
}
}
}
return true
})
c.JSON(200, rspObj)
}
func (this *mailMgr) CaMarkMail(c *gin.Context) {
hum := c.MustGet("hum").(common.Player)
mailIds := q5.StrSplit(c.DefaultQuery("mail_ids", ""), ",")
mails := []common.Mail{}
for _, str := range(mailIds) {
m := this.getMail(str)
if m != nil && m.IsValid(hum) {
q5.AppendSlice(&mails, m)
}
}
hum.MarkMails(mails)
rspObj := struct {
ErrCode int32 `json:"errcode"`
ErrMsg string `json:"errmsg"`
}{}
c.JSON(200, rspObj)
}
func (this *mailMgr) CaGetUnreadMailCnt(c *gin.Context) {
hum := c.MustGet("hum").(common.Player)
rspObj := struct {
ErrCode int32 `json:"errcode"`
ErrMsg string `json:"errmsg"`
UnreadMailCnt int32 `json:"unread_mail_cnt"`
}{}
this.traversePlayerMail(
hum,
func (m *mail) bool {
if m.IsValid(hum) && hum.IsUnread(m) {
rspObj.UnreadMailCnt++
return false
}
return true
})
c.JSON(200, rspObj)
}
func (this *mailMgr) CaGetAttachment(c *gin.Context) {
hum := c.MustGet("hum").(common.Player)
mailIds := q5.StrSplit(c.DefaultQuery("mail_ids", ""), ",")
mails := []common.Mail{}
for _, str := range(mailIds) {
m := this.getMail(str)
if m != nil && m.IsValid(hum) {
q5.AppendSlice(&mails, m)
}
}
hum.GetAttachment(mails, c)
}
func (this *mailMgr) CaDeleteMails(c *gin.Context) {
hum := c.MustGet("hum").(common.Player)
mailIds := q5.StrSplit(c.DefaultQuery("mail_ids", ""), ",")
mails := []common.Mail{}
for _, str := range(mailIds) {
m := this.getMail(str)
if m != nil && m.IsValid(hum) {
q5.AppendSlice(&mails, m)
}
}
hum.DeleteMails(mails)
rspObj := struct {
ErrCode int32 `json:"errcode"`
ErrMsg string `json:"errmsg"`
}{}
c.JSON(200, rspObj)
}
func (this *mailMgr) CaSendMail(c *gin.Context) {
rspObj := struct {
ErrCode int32 `json:"errcode"`
ErrMsg string `json:"errmsg"`
}{}
key := c.DefaultQuery("key", "")
if mt.Table.Config.GetById(0).GetGmSecretKey() != key {
rspObj.ErrCode = 2
rspObj.ErrMsg = "is not gm error"
c.JSON(200, rspObj)
return
}
if mt.Table.Config.GetById(0).GetGmOpen() == 0 {
rspObj.ErrCode = 3
rspObj.ErrMsg = "is not gm open"
c.JSON(200, rspObj)
return
}
reqJson := struct {
UniKey string `json:"unikey"`
Subject string `json:"subject"`
Content string `json:"content"`
Recipients []string`json:"recipients"`
SendTime int32 `json:"sendtime"`
ExpireTime int32 `json:"expiretime"`
Attachments []*common.AttachmentDto `json:"attachments"`
}{}
if err := c.ShouldBindJSON(&reqJson); err != nil {
c.JSON(200, gin.H{
"code": 1,
"message": err.Error(),
})
return
}
if reqJson.UniKey == "" {
rspObj.ErrCode = 500
rspObj.ErrMsg = "unikey param error"
c.JSON(200, rspObj)
return
}
/*
if reqJson.Subject == "" {
c.JSON(200, rspObj)
return
}*/
if reqJson.Content == "" {
rspObj.ErrCode = 500
rspObj.ErrMsg = "unikey param error"
c.JSON(200, rspObj)
return
}
if len(reqJson.Recipients) <= 0 {
rspObj.ErrCode = 500
rspObj.ErrMsg = "recipients param error"
c.JSON(200, rspObj)
return
}
if f5.GetApp().GetRealSeconds() >= q5.ToInt64(reqJson.ExpireTime) {
rspObj.ErrCode = 500
rspObj.ErrMsg = "expiretime param error"
c.JSON(200, rspObj)
return
}
for _, val := range reqJson.Attachments {
if val.ItemId == 0 {
rspObj.ErrCode = 500
rspObj.ErrMsg = "Attachments param error"
c.JSON(200, rspObj)
return
}
if val.ItemNum <= 0 || val.ItemNum > 10000 * 100 {
rspObj.ErrCode = 500
rspObj.ErrMsg = "Attachments param error"
c.JSON(200, rspObj)
return
}
}
mailId := f5.GetApp().NewLockNodeUuid()
nowTime := f5.GetApp().GetRealSeconds()
f5.GetGoStyleDb().Insert(
constant.MAIL_DB,
"t_mail",
[][]string{
{"mail_id", q5.ToString(mailId)},
{"mail_type", q5.ToString(jccommon.MAIL_TYPE_GROUP)},
{"unikey", reqJson.UniKey},
{"subject", reqJson.Subject},
{"content", reqJson.Content},
{"recipients", q5.EncodeJson(reqJson.Recipients)},
{"attachments", q5.EncodeJson(reqJson.Attachments)},
{"sendtime", q5.ToString(reqJson.SendTime)},
{"expiretime", q5.ToString(reqJson.ExpireTime)},
{"user_reg_start_time", q5.ToString(0)},
{"user_reg_end_time", q5.ToString(nowTime + 3600 * 24 * 365 * 10)},
{"createtime", q5.ToString(nowTime)},
{"modifytime", q5.ToString(nowTime)},
},
func (err error, lastInsertId int64, rowsAffected int64) {
if err != nil {
rspObj.ErrCode = 500
rspObj.ErrMsg = err.Error()
return
}
f5.GetGoStyleDb().Insert(
constant.MAIL_DB,
"t_event",
[][]string{
{"event_name", constant.EVENT_MAIL_UPDATE},
{"param1", q5.ToString(mailId)},
{"createtime", q5.ToString(nowTime)},
{"modifytime", q5.ToString(nowTime)},
},
func (err error, lastInsertId int64, rowsAffected int64) {
c.JSON(200, rspObj)
})
})
}
func (this *mailMgr) traversePlayerMail(hum common.Player, cb func(*mail) bool) {
stop := false
traversedMails := make(map[int64]*mail, 10)
traversFunc := func (k, v interface{}) bool {
m := v.(*mail)
if m.IsValid(hum) {
if _, ok := traversedMails[m.mailId]; ok {
return true
}
if !(cb(m)) {
stop = true
return false
}
traversedMails[m.mailId] = m
}
return true
}
this.wholeMails.Range(traversFunc)
if stop {
return
}
this.groupMails.Range(traversFunc)
if stop {
return
}
if p, ok := this.personalMails.Load(hum.GetAccountId()); ok {
(p.(*sync.Map)).Range(traversFunc)
}
}
func (this *mailMgr) removeMail(m *mail) {
this.idHash.Delete(m.mailId)
if m.isType(constant.MAIL_TYPE_ALL) {
this.wholeMails.Delete(m.mailId)
} else if m.isType(constant.MAIL_TYPE_GROUP) {
m.traverseUserGroup(
func (int64, *userGroup) bool {
this.groupMails.Delete(m.mailId)
return false
})
m.traverseRecipients(
func (accountId string) bool {
if p, ok := this.personalMails.Load(accountId); ok {
(p.(*sync.Map)).Delete(m.mailId)
}
return true
})
}
}
func (this *mailMgr) addMail(m *mail) {
this.idHash.Store(m.mailId, m)
if m.isType(constant.MAIL_TYPE_ALL) {
this.wholeMails.Store(m.mailId, m)
} else if m.isType(constant.MAIL_TYPE_GROUP) {
m.traverseUserGroup(
func (int64, *userGroup) bool {
this.groupMails.Store(m.mailId, m)
return false
})
m.traverseRecipients(
func (accountId string) bool {
if p, ok := this.personalMails.Load(accountId); ok {
(p.(*sync.Map)).Store(m.mailId, m)
} else {
p := new(sync.Map)
p.Store(m.mailId, m)
this.personalMails.Store(accountId, p)
}
return true
})
}
}
func (this *mailMgr) updateMail(m *mail) {
oldM := this.internalGetMail(m.mailId)
if oldM == nil {
this.addMail(m)
return
}
if oldM.mailType != m.mailType {
panic(fmt.Sprintf("updateMail mailType error"));
return
}
this.removeMail(oldM)
this.addMail(m)
}
func (this *mailMgr) addGroup(g *userGroup) {
this.groupHash.Store(g.groupId, g)
}
func (this *mailMgr) getGroup(groupId int64) *userGroup {
if p, ok := this.groupHash.Load(groupId); ok {
return p.(*userGroup)
} else {
return nil
}
}
func (this *mailMgr) getMail(mailId string) common.Mail {
if m := this.internalGetMail(q5.ToInt64(mailId)); m != nil {
return m
} else {
return nil
}
}
func (this *mailMgr) internalGetMail(mailId int64) *mail {
if p, ok := this.idHash.Load(mailId); ok {
return p.(*mail)
} else {
return nil
}
}
func (this *mailMgr) syncEvent() {
f5.GetGoStyleDb().SyncSelectCustomQuery(
constant.MAIL_DB,
"SELECT MAX(idx) FROM t_event;",
func(err error, ds *f5.DataSet) {
if err != nil {
panic("sync event error")
return
}
if ds.Next() {
this.lastSyncEventIdx = q5.ToInt64(ds.GetByIndex(0))
} else {
panic("sync event error")
}
});
f5.GetTimer().SetInterval(
1000 * 2,
func (e int32, args *q5.Args) {
if e == q5.TIMER_EXEC_EVENT {
this.pullEvent()
}
})
}
func (this *mailMgr) addDbEvent(e *dbEvent) {
this.eventQueue.Push(&e.entry)
}
func (this *mailMgr) updateDbEvent() {
if this.eventQueue.IsEmpty() {
return
}
if this.procingEvent {
return
}
this.eventQueue.Fetch()
for !this.eventQueue.WorkList.Empty() {
e := this.eventQueue.WorkList.FirstEntry().(*dbEvent)
if e.eventName == constant.EVENT_MAIL_UPDATE {
this.procMailUpdate(e)
break
} else if e.eventName == constant.EVENT_UPSER_GROUP_UPDATE {
this.procGroupUpdate(e)
break
} else {
e.entry.DelInit()
}
}
}
func (this *mailMgr) pullEvent() {
if this.pullingEvent {
return
}
this.pullingEvent = true
f5.GetJsStyleDb().SelectCustomQuery(
constant.MAIL_DB,
fmt.Sprintf("SELECT * FROM t_event WHERE idx > %d;", this.lastSyncEventIdx),
func(err error, ds *f5.DataSet) {
if err != nil {
this.pullingEvent = false
return
}
for ds.Next() {
e := new(dbEvent)
e.idx = q5.ToInt64(ds.GetByName("idx"))
e.eventName = ds.GetByName("event_name")
e.param1 = ds.GetByName("param1")
e.entry.Init(e)
_mailMgr.addDbEvent(e)
this.lastSyncEventIdx = q5.ToInt64(ds.GetByName("idx"))
}
this.pullingEvent = false
})
}
func (this* mailMgr) HasTask() bool {
return false
}
func (this* mailMgr) procMailUpdate(e *dbEvent) {
this.procingEvent = true
f5.GetJsStyleDb().OrmSelectOne(
constant.MAIL_DB,
"t_mail",
[][]string{
{"mail_id", e.param1},
},
func (err error, ds *f5.DataSet) {
this.procingEvent = false
if err != nil {
return
}
if ds.Next() {
p := newMail()
p.loadFromDb(ds)
this.updateMail(p)
} else {
panic(fmt.Sprintf("procMailUpdate error:%s", err));
}
e.entry.DelInit()
this.updateDbEvent()
})
}
func (this* mailMgr) procGroupUpdate(e *dbEvent) {
this.procingEvent = true
fetchMemberFunc := func () {
f5.GetJsStyleDb().OrmSelect(
constant.MAIL_DB,
"t_member",
[][]string{
{"group_id", e.param1},
{"deleted", "0"},
},
func (err error, ds *f5.DataSet) {
this.procingEvent = false
if err != nil {
return
}
groupId := q5.ToInt64(ds.GetByName("group_id"))
g := this.getGroup(groupId)
if g == nil {
p := newUserGroup()
p.groupId = groupId
this.addGroup(g)
} else {
g.userHash = new(sync.Map)
}
if ds.Next() {
memberId := ds.GetByName("member_id")
g.userHash.Store(memberId, q5.ToInt64(ds.GetByName("idx")))
return
} else {
panic(fmt.Sprintf("procGroupUpdate1 error:%s", err));
}
})
}
f5.GetJsStyleDb().OrmSelectOne(
constant.MAIL_DB,
"t_group",
[][]string{
{"group_id", e.param1},
},
func (err error, ds *f5.DataSet) {
if err != nil {
this.procingEvent = false
return
}
if ds.Next() {
deleted := q5.ToInt32(ds.GetByName("deleted"))
if deleted != 0 {
this.procingEvent = false
this.groupHash.Delete(q5.ToInt64(e.param1))
return
}
fetchMemberFunc()
return
} else {
this.procingEvent = false
panic(fmt.Sprintf("procGroupUpdate1 error:%s", err));
}
})
}
func newUserGroup() *userGroup {
g := new(userGroup)
g.userHash = new(sync.Map)
return g
}