90 lines
2.4 KiB
Go
90 lines
2.4 KiB
Go
package engine
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
|
|
"aiaa-notification-service/internal/adapter"
|
|
"aiaa-notification-service/internal/cache"
|
|
"aiaa-notification-service/internal/model"
|
|
"aiaa-notification-service/internal/retry"
|
|
"aiaa-notification-service/internal/store"
|
|
)
|
|
|
|
type Router struct {
|
|
store *store.Store
|
|
cache *cache.Cache
|
|
senderFactory func(channelType string) (adapter.ChannelSender, error)
|
|
}
|
|
|
|
func NewRouter(s *store.Store, c *cache.Cache, sf func(channelType string) (adapter.ChannelSender, error)) *Router {
|
|
return &Router{store: s, cache: c, senderFactory: sf}
|
|
}
|
|
|
|
// Route sends content to all enabled channels for the rule. Returns immediately, sends async.
|
|
func (r *Router) Route(ctx context.Context, rule *model.Rule, title, content string) []string {
|
|
channels, err := r.getChannels(ctx, rule.ID)
|
|
if err != nil {
|
|
slog.Error("get channels for rule", "rule_id", rule.ID, "error", err)
|
|
return nil
|
|
}
|
|
|
|
channelNames := make([]string, 0, len(channels))
|
|
for _, ch := range channels {
|
|
channelNames = append(channelNames, fmt.Sprintf("%s:%d", ch.Type, ch.ID))
|
|
sender, err := r.senderFactory(ch.Type)
|
|
if err != nil {
|
|
slog.Error("create sender", "type", ch.Type, "error", err)
|
|
continue
|
|
}
|
|
|
|
go func(ch cache.CachedChannel, s adapter.ChannelSender) {
|
|
var cfg json.RawMessage
|
|
if ch.Config != nil {
|
|
cfg = *ch.Config
|
|
}
|
|
r := retry.DefaultRetrier()
|
|
if err := r.Do(context.Background(), func() error {
|
|
return s.Send(title, content, cfg)
|
|
}); err != nil {
|
|
slog.Error("send failed after retries", "channel_type", ch.Type, "channel_id", ch.ID, "error", err)
|
|
} else {
|
|
slog.Info("sent", "channel_type", ch.Type, "channel_id", ch.ID)
|
|
}
|
|
}(ch, sender)
|
|
}
|
|
|
|
return channelNames
|
|
}
|
|
|
|
func (r *Router) getChannels(ctx context.Context, ruleID int) ([]cache.CachedChannel, error) {
|
|
if r.cache != nil {
|
|
chs, err := r.cache.GetChannels(ctx, ruleID)
|
|
if err == nil {
|
|
return chs, nil
|
|
}
|
|
}
|
|
|
|
rcs, err := r.store.GetRuleChannels(ctx, ruleID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var cached []cache.CachedChannel
|
|
for _, rc := range rcs {
|
|
ch, err := r.store.GetChannel(ctx, rc.ChannelID)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
cc := cache.CachedChannel{ID: ch.ID, Type: ch.Type, Config: ch.Config}
|
|
cached = append(cached, cc)
|
|
}
|
|
|
|
if r.cache != nil {
|
|
_ = r.cache.SetChannels(ctx, ruleID, cached)
|
|
}
|
|
return cached, nil
|
|
}
|