93 lines
2.4 KiB
Go
93 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/store"
|
|
)
|
|
|
|
type SendResult struct {
|
|
ChannelType string `json:"channel_type"`
|
|
ChannelID int `json:"channel_id"`
|
|
Success bool `json:"success"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
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
|
|
}
|
|
if err := s.Send(title, content, cfg); err != nil {
|
|
slog.Error("send failed", "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
|
|
}
|