HyperNet/Messaging
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

♻️ Rebuilt cache with nexus cache

Browse Source
This commit is contained in:
LittleSheep 2025-03-29 15:02:04 +08:00
parent c24ed1e7e6
commit e916eb2395
8 changed files with 74 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
go.mod
View File

@ -7,9 +7,6 @@ require (
git.solsynth.dev/hypernet/paperclip v0.0.0-20250310151112-1d866f317f47
git.solsynth.dev/hypernet/passport v0.0.0-20250315083747-32e91e26013c
git.solsynth.dev/hypernet/pusher v0.0.0-20250216145944-5fb769823a88
github.com/dgraph-io/ristretto v0.2.0
github.com/eko/gocache/lib/v4 v4.2.0
github.com/eko/gocache/store/ristretto/v4 v4.2.2
github.com/fatih/color v1.18.0
github.com/go-playground/validator/v10 v10.22.1
github.com/gofiber/fiber/v2 v2.52.6
@ -37,9 +34,10 @@ require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/eapache/channels v1.1.0 // indirect
github.com/eapache/queue v1.1.0 // indirect
github.com/eko/gocache/lib/v4 v4.2.0 // indirect
github.com/eko/gocache/store/redis/v4 v4.2.2 // indirect
github.com/frostbyte73/core v0.0.10 // indirect
github.com/fsnotify/fsnotify v1.8.0 // indirect
github.com/gabriel-vasile/mimetype v1.4.3 // indirect
@ -97,7 +95,6 @@ require (
github.com/pion/transport/v2 v2.2.4 // indirect
github.com/pion/turn/v2 v2.1.3 // indirect
github.com/pion/webrtc/v3 v3.2.28 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/prometheus/client_golang v1.19.0 // indirect
github.com/prometheus/client_model v0.6.1 // indirect
@ -119,8 +116,6 @@ require (
github.com/twitchtv/twirp v8.1.3+incompatible // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/valyala/fasthttp v1.59.0 // indirect
github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
github.com/zeebo/xxh3 v1.0.2 // indirect
go.uber.org/atomic v1.11.0 // indirect
go.uber.org/mock v0.4.0 // indirect

14
go.sum
View File

@ -27,22 +27,16 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dgraph-io/ristretto v0.2.0 h1:XAfl+7cmoUDWW/2Lx8TGZQjjxIQ2Ley9DSf52dru4WE=
github.com/dgraph-io/ristretto v0.2.0/go.mod h1:8uBHCU/PBV4Ag0CJrP47b9Ofby5dqWNh4FicAdoqFNU=
github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 h1:fAjc9m62+UWV/WAFKLNi6ZS0675eEUC9y3AlwSbQu1Y=
github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/eapache/channels v1.1.0 h1:F1taHcn7/F0i8DYqKXJnyhJcVpp2kgFcNePxXtnyu4k=
github.com/eapache/channels v1.1.0/go.mod h1:jMm2qB5Ubtg9zLd+inMZd2/NUvXgzmWXsDaLyQIGfH0=
github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
github.com/eko/gocache/lib/v4 v4.2.0 h1:MNykyi5Xw+5Wu3+PUrvtOCaKSZM1nUSVftbzmeC7Yuw=
github.com/eko/gocache/lib/v4 v4.2.0/go.mod h1:7ViVmbU+CzDHzRpmB4SXKyyzyuJ8A3UW3/cszpcqB4M=
github.com/eko/gocache/store/ristretto/v4 v4.2.2 h1:lXFzoZ5ck6Gy6ON7f5DHSkNt122qN7KoroCVgVwF7oo=
github.com/eko/gocache/store/ristretto/v4 v4.2.2/go.mod h1:uIvBVJzqRepr5L0RsbkfQ2iYfbyos2fuji/s4yM+aUM=
github.com/eko/gocache/store/redis/v4 v4.2.2 h1:Thw31fzGuH3WzJywsdbMivOmP550D6JS7GDHhvCJPA0=
github.com/eko/gocache/store/redis/v4 v4.2.2/go.mod h1:LaTxLKx9TG/YUEybQvPMij++D7PBTIJ4+pzvk0ykz0w=
github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
@ -315,10 +309,6 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fasthttp v1.59.0 h1:Qu0qYHfXvPk1mSLNqcFtEk6DpxgA26hy6bmydotDpRI=
github.com/valyala/fasthttp v1.59.0/go.mod h1:GTxNb9Bc6r2a9D0TWNSPwDz78UxnTGBViY3xZNEqyYU=
github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

24
pkg/internal/cache/store.go vendored
View File

@ -1,24 +0,0 @@
package cache
import (
"github.com/dgraph-io/ristretto"
"github.com/eko/gocache/lib/v4/store"
ristrettoCache "github.com/eko/gocache/store/ristretto/v4"
)
var S store.StoreInterface
func NewStore() error {
ristretto, err := ristretto.NewCache(&ristretto.Config{
NumCounters: 1000,
MaxCost: 100,
BufferItems: 64,
})
if err != nil {
return err
}
S = ristrettoCache.NewRistretto(ristretto)
return nil
}

18
pkg/internal/gap/server.go
View File

@ -2,19 +2,25 @@ package gap
import (
"fmt"
"strings"
"time"
"git.solsynth.dev/hypernet/nexus/pkg/nex"
"git.solsynth.dev/hypernet/nexus/pkg/nex/cachekit"
"git.solsynth.dev/hypernet/nexus/pkg/proto"
"git.solsynth.dev/hypernet/pusher/pkg/pushkit/pushcon"
"github.com/samber/lo"
"strings"
"github.com/rs/zerolog/log"
"github.com/spf13/viper"
)
var Nx *nex.Conn
var Px *pushcon.Conn
var (
Nx *nex.Conn
Px *pushcon.Conn
Ca *cachekit.CaConn
)
func InitializeToNexus() error {
grpcBind := strings.SplitN(viper.GetString("grpc_bind"), ":", 2)
@ -47,6 +53,10 @@ func InitializeToNexus() error {
return fmt.Errorf("error during initialize pushcon: %v", err)
}
return err
Ca, err = cachekit.NewCaConn(Nx, 3*time.Second)
if err != nil {
return fmt.Errorf("error during initialize cachekit: %v", err)
}
return err
}

48
pkg/internal/services/channel_members.go
View File

@ -1,17 +1,13 @@
package services
import (
"context"
"errors"
"fmt"
localCache "git.solsynth.dev/hypernet/messaging/pkg/internal/cache"
"git.solsynth.dev/hypernet/messaging/pkg/internal/gap"
"git.solsynth.dev/hypernet/nexus/pkg/nex/cachekit"
"git.solsynth.dev/hypernet/passport/pkg/authkit"
authm "git.solsynth.dev/hypernet/passport/pkg/authkit/models"
"github.com/eko/gocache/lib/v4/cache"
"github.com/eko/gocache/lib/v4/marshaler"
"github.com/eko/gocache/lib/v4/store"
"gorm.io/gorm"
"git.solsynth.dev/hypernet/messaging/pkg/internal/database"
@ -81,16 +77,10 @@ func AddChannelMember(user authm.Account, target models.Channel) error {
err := database.C.Save(&member).Error
if err == nil {
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
ctx := context.Background()
_ = marshal.Invalidate(
ctx,
store.WithInvalidateTags([]string{
fmt.Sprintf("channel#%d", target.ID),
fmt.Sprintf("user#%d", user.ID),
}),
cachekit.DeleteByTags(
gap.Ca,
fmt.Sprintf("channel#%d", target.ID),
fmt.Sprintf("user#%d", user.ID),
)
}
@ -101,16 +91,10 @@ func EditChannelMember(membership models.ChannelMember) (models.ChannelMember, e
if err := database.C.Save(&membership).Error; err != nil {
return membership, err
} else {
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
contx := context.Background()
_ = marshal.Invalidate(
contx,
store.WithInvalidateTags([]string{
fmt.Sprintf("channel#%d", membership.ChannelID),
fmt.Sprintf("user#%d", membership.AccountID),
}),
cachekit.DeleteByTags(
gap.Ca,
fmt.Sprintf("channel#%d", membership.ChannelID),
fmt.Sprintf("user#%d", membership.AccountID),
)
}
@ -121,16 +105,10 @@ func RemoveChannelMember(member models.ChannelMember, target models.Channel) err
if err := database.C.Delete(&member).Error; err == nil {
database.C.Where("sender_id = ?").Delete(&models.Event{})
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
ctx := context.Background()
_ = marshal.Invalidate(
ctx,
store.WithInvalidateTags([]string{
fmt.Sprintf("channel#%d", target.ID),
fmt.Sprintf("user#%d", target.AccountID),
}),
cachekit.DeleteByTags(
gap.Ca,
fmt.Sprintf("channel#%d", target.ID),
fmt.Sprintf("user#%d", member.AccountID),
)
return nil

72
pkg/internal/services/channels.go
View File

@ -1,17 +1,15 @@
package services
import (
"context"
"fmt"
"regexp"
"time"
localCache "git.solsynth.dev/hypernet/messaging/pkg/internal/cache"
"git.solsynth.dev/hypernet/nexus/pkg/nex/cachekit"
authm "git.solsynth.dev/hypernet/passport/pkg/authkit/models"
"github.com/eko/gocache/lib/v4/cache"
"github.com/eko/gocache/lib/v4/marshaler"
"github.com/eko/gocache/lib/v4/store"
"git.solsynth.dev/hypernet/messaging/pkg/internal/database"
"git.solsynth.dev/hypernet/messaging/pkg/internal/gap"
"git.solsynth.dev/hypernet/messaging/pkg/internal/models"
"github.com/samber/lo"
"github.com/spf13/viper"
@ -23,7 +21,7 @@ type channelIdentityCacheEntry struct {
ChannelMember models.ChannelMember `json:"channel_member"`
}
func GetChannelIdentityCacheKey(channel string, user uint, realm ...uint) string {
func KgChannelIdentityCache(channel string, user uint, realm ...uint) string {
if len(realm) > 0 {
return fmt.Sprintf("channel-identity-%s#%d@%d", channel, user, realm)
} else {
@ -31,18 +29,16 @@ func GetChannelIdentityCacheKey(channel string, user uint, realm ...uint) string
}
}
func CacheChannelIdentityCache(channel models.Channel, member models.ChannelMember, user uint, realm ...uint) {
key := GetChannelIdentityCacheKey(channel.Alias, user, realm...)
func CacheChannelIdentity(channel models.Channel, member models.ChannelMember, user uint, realm ...uint) {
key := KgChannelIdentityCache(channel.Alias, user, realm...)
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
contx := context.Background()
_ = marshal.Set(
contx,
cachekit.Set(
gap.Ca,
key,
channelIdentityCacheEntry{channel, member},
store.WithTags([]string{"channel-identity", fmt.Sprintf("channel#%d", channel.ID), fmt.Sprintf("user#%d", user)}),
60*time.Minute,
fmt.Sprintf("channel#%d", channel.ID),
fmt.Sprintf("user#%d", user),
)
}
@ -60,37 +56,37 @@ func GetChannelIdentityWithID(id uint, user uint) (models.Channel, models.Channe
}
func GetChannelIdentity(alias string, user uint, realm ...authm.Realm) (models.Channel, models.ChannelMember, error) {
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
contx := context.Background()
var err error
var channel models.Channel
var member models.ChannelMember
hitCache := false
if len(realm) > 0 {
if val, err := marshal.Get(contx, GetChannelIdentityCacheKey(alias, user, realm[0].ID), new(channelIdentityCacheEntry)); err == nil {
entry := val.(*channelIdentityCacheEntry)
channel = entry.Channel
member = entry.ChannelMember
if val, err := cachekit.Get[channelIdentityCacheEntry](
gap.Ca,
KgChannelIdentityCache(alias, user, realm[0].ID),
); err == nil {
channel = val.Channel
member = val.ChannelMember
hitCache = true
}
} else {
if val, err := marshal.Get(contx, GetChannelIdentityCacheKey(alias, user), new(channelIdentityCacheEntry)); err == nil {
entry := val.(*channelIdentityCacheEntry)
channel = entry.Channel
member = entry.ChannelMember
if val, err := cachekit.Get[channelIdentityCacheEntry](
gap.Ca,
KgChannelIdentityCache(alias, user),
); err == nil {
channel = val.Channel
member = val.ChannelMember
hitCache = true
}
}
if !hitCache {
if len(realm) > 0 {
channel, member, err = GetAvailableChannelWithAlias(alias, user, realm[0].ID)
CacheChannelIdentityCache(channel, member, user, realm[0].ID)
CacheChannelIdentity(channel, member, user, realm[0].ID)
} else {
channel, member, err = GetAvailableChannelWithAlias(alias, user)
CacheChannelIdentityCache(channel, member, user)
CacheChannelIdentity(channel, member, user)
}
}
@ -282,14 +278,7 @@ func EditChannel(channel models.Channel) (models.Channel, error) {
err := database.C.Save(&channel).Error
if err == nil {
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
contx := context.Background()
_ = marshal.Invalidate(
contx,
store.WithInvalidateTags([]string{fmt.Sprintf("channel#%d", channel.ID)}),
)
cachekit.DeleteByTags(gap.Ca, fmt.Sprintf("channel#%d", channel.ID))
}
return channel, err
@ -302,14 +291,7 @@ func DeleteChannel(channel models.Channel) error {
database.C.Where("channel_id = ?", channel.ID).Delete(&models.Event{})
database.C.Where("channel_id = ?", channel.ID).Delete(&models.ChannelMember{})
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
contx := context.Background()
_ = marshal.Invalidate(
contx,
store.WithInvalidateTags([]string{fmt.Sprintf("channel#%d", channel.ID)}),
)
cachekit.DeleteByTags(gap.Ca, fmt.Sprintf("channel#%d", channel.ID))
return nil
} else {

35
pkg/internal/services/status.go
View File

@ -1,16 +1,14 @@
package services
import (
"context"
"fmt"
"time"
localCache "git.solsynth.dev/hypernet/messaging/pkg/internal/cache"
"git.solsynth.dev/hypernet/messaging/pkg/internal/database"
"git.solsynth.dev/hypernet/messaging/pkg/internal/gap"
"git.solsynth.dev/hypernet/messaging/pkg/internal/models"
"git.solsynth.dev/hypernet/nexus/pkg/nex"
"github.com/eko/gocache/lib/v4/cache"
"github.com/eko/gocache/lib/v4/marshaler"
"github.com/eko/gocache/lib/v4/store"
"git.solsynth.dev/hypernet/nexus/pkg/nex/cachekit"
"github.com/samber/lo"
"github.com/spf13/viper"
)
@ -20,23 +18,21 @@ type statusQueryCacheEntry struct {
Data any
}
func GetTypingStatusQueryCacheKey(channelId uint, userId uint) string {
return fmt.Sprintf("typing-status-query#%d;%d", channelId, userId)
func KgTypingStatusCache(channelId uint, userId uint) string {
return fmt.Sprintf("chat-typing-status#%d@%d", userId, channelId)
}
func SetTypingStatus(channelId uint, userId uint) error {
var broadcastTarget []uint64
var data any
cacheManager := cache.New[any](localCache.S)
marshal := marshaler.New(cacheManager)
contx := context.Background()
hitCache := false
if val, err := marshal.Get(contx, GetTypingStatusQueryCacheKey(channelId, userId), new(statusQueryCacheEntry)); err == nil {
entry := val.(*statusQueryCacheEntry)
broadcastTarget = entry.Target
data = entry.Data
if val, err := cachekit.Get[statusQueryCacheEntry](
gap.Ca,
KgTypingStatusCache(channelId, userId),
); err == nil {
broadcastTarget = val.Target
data = val.Data
hitCache = true
}
@ -69,11 +65,12 @@ func SetTypingStatus(channelId uint, userId uint) error {
}
// Cache queries
_ = marshal.Set(
contx,
GetTypingStatusQueryCacheKey(channelId, userId),
cachekit.Set(
gap.Ca,
KgTypingStatusCache(channelId, userId),
statusQueryCacheEntry{broadcastTarget, data},
store.WithTags([]string{"typing-status-query", fmt.Sprintf("channel#%d", channelId), fmt.Sprintf("user#%d", userId)}),
60*time.Minute,
fmt.Sprintf("channel#%d", channelId),
)
}

6
pkg/main.go
View File

@ -18,7 +18,6 @@ import (
"git.solsynth.dev/hypernet/messaging/pkg/internal/web"
pkg "git.solsynth.dev/hypernet/messaging/pkg/internal"
"git.solsynth.dev/hypernet/messaging/pkg/internal/cache"
"git.solsynth.dev/hypernet/messaging/pkg/internal/database"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
@ -68,11 +67,6 @@ func main() {
log.Fatal().Err(err).Msg("An error occurred when running database auto migration.")
}
// Initialize cache
if err := cache.NewStore(); err != nil {
log.Fatal().Err(err).Msg("An error occurred when initializing cache.")
}
// Connect other services
services.SetupLiveKit()