Skip to content

Commit 92e6863

Browse files
MuZhou233MuZhou233
committed
feat: add Netzach module
1 parent fa38035 commit 92e6863

94 files changed

Lines changed: 16177 additions & 375 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎app/sephirah/cmd/sephirah/main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ func main() {
4242
var bc conf.Sephirah
4343
appSettings.LoadConfig(&bc)
4444

45-
app, cleanup, err := wireApp(bc.Server, bc.Data, bc.Auth, bc.Mq, appSettings)
45+
app, cleanup, err := wireApp(bc.Server, bc.Data, bc.Auth, bc.Mq, bc.Cache, appSettings)
4646
if err != nil {
4747
panic(err)
4848
}

‎app/sephirah/cmd/sephirah/wire.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"github.com/tuihub/librarian/internal/conf"
1313
"github.com/tuihub/librarian/internal/lib/libapp"
1414
"github.com/tuihub/librarian/internal/lib/libauth"
15+
"github.com/tuihub/librarian/internal/lib/libcache"
1516
"github.com/tuihub/librarian/internal/lib/libcron"
1617
"github.com/tuihub/librarian/internal/lib/libmq"
1718
"github.com/tuihub/librarian/internal/server"
@@ -26,6 +27,7 @@ func wireApp(
2627
*conf.Sephirah_Data,
2728
*conf.Auth,
2829
*conf.MQ,
30+
*conf.Cache,
2931
*libapp.Settings,
3032
) (*kratos.App, func(), error) {
3133
panic(wire.Build(
@@ -37,6 +39,7 @@ func wireApp(
3739
libauth.ProviderSet,
3840
libmq.ProviderSet,
3941
libcron.ProviderSet,
42+
libcache.ProviderSet,
4043
newApp,
4144
))
4245
}

‎app/sephirah/cmd/sephirah/wire_gen.go‎

Lines changed: 20 additions & 5 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎app/sephirah/internal/biz/biz.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"github.com/tuihub/librarian/app/sephirah/internal/biz/bizangela"
55
"github.com/tuihub/librarian/app/sephirah/internal/biz/bizbinah"
66
"github.com/tuihub/librarian/app/sephirah/internal/biz/bizgebura"
7+
"github.com/tuihub/librarian/app/sephirah/internal/biz/biznetzach"
78
"github.com/tuihub/librarian/app/sephirah/internal/biz/biztiphereth"
89
"github.com/tuihub/librarian/app/sephirah/internal/biz/bizyesod"
910

@@ -18,4 +19,5 @@ var ProviderSet = wire.NewSet(
1819
bizbinah.NewBinah,
1920
bizbinah.NewCallbackControl,
2021
bizyesod.NewYesod,
22+
biznetzach.NewNetzach,
2123
)

‎app/sephirah/internal/biz/bizangela/angela.go‎

Lines changed: 102 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package bizangela
22

33
import (
44
"context"
5+
"errors"
56
"net/url"
67
"strconv"
78
"time"
@@ -12,9 +13,11 @@ import (
1213
"github.com/tuihub/librarian/app/sephirah/internal/model/converter"
1314
"github.com/tuihub/librarian/app/sephirah/internal/model/modelangela"
1415
"github.com/tuihub/librarian/app/sephirah/internal/model/modelgebura"
16+
"github.com/tuihub/librarian/app/sephirah/internal/model/modelnetzach"
1517
"github.com/tuihub/librarian/app/sephirah/internal/model/modeltiphereth"
1618
"github.com/tuihub/librarian/app/sephirah/internal/model/modelyesod"
1719
"github.com/tuihub/librarian/internal/lib/libapp"
20+
"github.com/tuihub/librarian/internal/lib/libcache"
1821
"github.com/tuihub/librarian/internal/lib/libmq"
1922
"github.com/tuihub/librarian/internal/model"
2023
"github.com/tuihub/librarian/internal/model/modelfeed"
@@ -24,6 +27,7 @@ import (
2427
librarian "github.com/tuihub/protos/pkg/librarian/v1"
2528

2629
"github.com/google/wire"
30+
"golang.org/x/exp/slices"
2731
)
2832

2933
var ProviderSet = wire.NewSet(
@@ -33,6 +37,11 @@ var ProviderSet = wire.NewSet(
3337
NewPullSteamAccountAppRelationTopic,
3438
NewPullSteamAppTopic,
3539
NewPullFeedTopic,
40+
NewNotifyRouterTopic,
41+
NewNotifyPushTopic,
42+
NewFeedToNotifyFlowMap,
43+
NewNotifyFlowCache,
44+
NewNotifyTargetCache,
3645
)
3746

3847
type Angela struct {
@@ -71,6 +80,8 @@ func NewAngela(
7180
pullSteamAccountAppRelation *libmq.Topic[modelangela.PullSteamAccountAppRelation],
7281
pullSteamApp *libmq.Topic[modelangela.PullSteamApp],
7382
pullFeed *libmq.Topic[modelyesod.PullFeed],
83+
notifyRouter *libmq.Topic[modelangela.NotifyRouter],
84+
notifyPush *libmq.Topic[modelangela.NotifyPush],
7485
) (*Angela, error) {
7586
if err := mq.RegisterTopic(pullAccount); err != nil {
7687
return nil, err
@@ -84,6 +95,12 @@ func NewAngela(
8495
if err := mq.RegisterTopic(pullFeed); err != nil {
8596
return nil, err
8697
}
98+
if err := mq.RegisterTopic(notifyRouter); err != nil {
99+
return nil, err
100+
}
101+
if err := mq.RegisterTopic(notifyPush); err != nil {
102+
return nil, err
103+
}
87104
return &Angela{
88105
mq: mq,
89106
}, nil
@@ -279,8 +296,9 @@ func NewPullSteamAppTopic(
279296
)
280297
}
281298

282-
func NewPullFeedTopic(
299+
func NewPullFeedTopic( //nolint:gocognit // TODO
283300
a *AngelaBase,
301+
notify *libmq.Topic[modelangela.NotifyRouter],
284302
) *libmq.Topic[modelyesod.PullFeed] {
285303
return libmq.NewTopic[modelyesod.PullFeed](
286304
"PullFeed",
@@ -321,7 +339,89 @@ func NewPullFeedTopic(
321339
item.PublishedParsed = &t
322340
}
323341
}
324-
return a.y.UpsertFeedItems(ctx, feed.Items, feed.ID)
342+
newItemGUIDs, err := a.y.UpsertFeedItems(ctx, feed.Items, feed.ID)
343+
if err != nil {
344+
return err
345+
}
346+
newItems := make([]*modelfeed.Item, 0, len(newItemGUIDs))
347+
for _, item := range feed.Items {
348+
if slices.Contains(newItemGUIDs, item.GUID) {
349+
newItems = append(newItems, item)
350+
}
351+
}
352+
err = notify.Publish(ctx, modelangela.NotifyRouter{
353+
FeedID: feed.ID,
354+
Messages: newItems,
355+
})
356+
if err != nil {
357+
return err
358+
}
359+
return nil
360+
},
361+
)
362+
}
363+
364+
func NewNotifyRouterTopic(
365+
a *AngelaBase,
366+
flowMap *libcache.Map[model.InternalID, modelnetzach.NotifyFlow],
367+
feedToFlowMap *libcache.Map[model.InternalID, modelangela.FeedToNotifyFlowValue],
368+
push *libmq.Topic[modelangela.NotifyPush],
369+
) *libmq.Topic[modelangela.NotifyRouter] {
370+
return libmq.NewTopic[modelangela.NotifyRouter](
371+
"NotifyRouter",
372+
func(ctx context.Context, r *modelangela.NotifyRouter) error {
373+
flowIDs, err := feedToFlowMap.GetWithFallBack(ctx, r.FeedID, nil)
374+
if err != nil {
375+
return err
376+
}
377+
if flowIDs == nil {
378+
return errors.New("nil result from feedToFlowMap")
379+
}
380+
for _, flowID := range *flowIDs {
381+
var flow *modelnetzach.NotifyFlow
382+
flow, err = flowMap.GetWithFallBack(ctx, flowID, nil)
383+
if err != nil {
384+
return err
385+
}
386+
for _, target := range flow.Targets {
387+
if target == nil {
388+
continue
389+
}
390+
err = push.Publish(ctx, modelangela.NotifyPush{
391+
Target: *target,
392+
Messages: r.Messages,
393+
})
394+
if err != nil {
395+
return err
396+
}
397+
}
398+
}
399+
return nil
400+
},
401+
)
402+
}
403+
404+
func NewNotifyPushTopic(
405+
a *AngelaBase,
406+
targetMap *libcache.Map[model.InternalID, modelnetzach.NotifyTarget],
407+
) *libmq.Topic[modelangela.NotifyPush] {
408+
return libmq.NewTopic[modelangela.NotifyPush](
409+
"NotifyPush",
410+
func(ctx context.Context, p *modelangela.NotifyPush) error {
411+
target, err := targetMap.GetWithFallBack(ctx, p.Target.TargetID, nil)
412+
if err != nil {
413+
return err
414+
}
415+
_, err = a.porter.PushFeedItems(ctx, &porter.PushFeedItemsRequest{
416+
Destination: converter.ToPBFeedDestination(target.Type),
417+
ChannelId: p.Target.ChannelID,
418+
Items: converter.ToPBFeedItemList(p.Messages),
419+
Token: target.Token,
420+
})
421+
if err != nil {
422+
return err
423+
}
424+
return nil
325425
},
326426
)
327427
}

‎app/sephirah/internal/biz/bizangela/notify.go‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
package bizangela
2+
3+
import (
4+
"context"
5+
"strconv"
6+
7+
"github.com/tuihub/librarian/app/sephirah/internal/biz/biznetzach"
8+
"github.com/tuihub/librarian/app/sephirah/internal/model/modelangela"
9+
"github.com/tuihub/librarian/app/sephirah/internal/model/modelnetzach"
10+
"github.com/tuihub/librarian/internal/lib/libcache"
11+
"github.com/tuihub/librarian/internal/lib/libtime"
12+
"github.com/tuihub/librarian/internal/model"
13+
)
14+
15+
// NewFeedToNotifyFlowMap Cache-Aside Pattern.
16+
func NewFeedToNotifyFlowMap(
17+
n biznetzach.NetzachRepo,
18+
store libcache.Store,
19+
) *libcache.Map[model.InternalID, modelangela.FeedToNotifyFlowValue] {
20+
return libcache.NewMap[model.InternalID, modelangela.FeedToNotifyFlowValue](
21+
store,
22+
"FeedToNotifyFlow",
23+
func(k model.InternalID) string {
24+
return strconv.FormatInt(int64(k), 10)
25+
},
26+
func(ctx context.Context, id model.InternalID) (*modelangela.FeedToNotifyFlowValue, error) {
27+
res, err := n.GetNotifyFlowIDsWithFeed(ctx, id)
28+
if err != nil {
29+
return nil, err
30+
}
31+
return (*modelangela.FeedToNotifyFlowValue)(&res), nil
32+
},
33+
libcache.WithExpiration(libtime.SevenDays),
34+
)
35+
}
36+
37+
// NewNotifyFlowCache Cache-Aside Pattern.
38+
func NewNotifyFlowCache(
39+
n biznetzach.NetzachRepo,
40+
store libcache.Store,
41+
) *libcache.Map[model.InternalID, modelnetzach.NotifyFlow] {
42+
return libcache.NewMap[model.InternalID, modelnetzach.NotifyFlow](
43+
store,
44+
"NotifyFlows",
45+
func(k model.InternalID) string {
46+
return strconv.FormatInt(int64(k), 10)
47+
},
48+
func(ctx context.Context, id model.InternalID) (*modelnetzach.NotifyFlow, error) {
49+
res, err := n.GetNotifyFlow(ctx, id)
50+
if err != nil {
51+
return nil, err
52+
}
53+
return res, nil
54+
},
55+
libcache.WithExpiration(libtime.SevenDays),
56+
)
57+
}
58+
59+
// NewNotifyTargetCache Cache-Aside Pattern.
60+
func NewNotifyTargetCache(
61+
n biznetzach.NetzachRepo,
62+
store libcache.Store,
63+
) *libcache.Map[model.InternalID, modelnetzach.NotifyTarget] {
64+
return libcache.NewMap[model.InternalID, modelnetzach.NotifyTarget](
65+
store,
66+
"NotifyTargets",
67+
func(k model.InternalID) string {
68+
return strconv.FormatInt(int64(k), 10)
69+
},
70+
func(ctx context.Context, id model.InternalID) (*modelnetzach.NotifyTarget, error) {
71+
res, err := n.GetNotifyTarget(ctx, id)
72+
if err != nil {
73+
return nil, err
74+
}
75+
return res, nil
76+
},
77+
libcache.WithExpiration(libtime.SevenDays),
78+
)
79+
}

0 commit comments

Comments
 (0)