diff --git a/internal/infra/notify/dispatcher.go b/internal/infra/notify/dispatcher.go new file mode 100644 index 0000000..74b03d9 --- /dev/null +++ b/internal/infra/notify/dispatcher.go @@ -0,0 +1,63 @@ +// Package notify 内的 dispatcher.go 实现通知分发的核心:先同步落库,再 +// 异步把消息广播到所有已注册的 Notifier。落库失败立即返回错误,外部投递 +// 不会启动;外部投递失败仅记日志,不影响调用方与其它通道。 +package notify + +import ( + "context" + "log/slog" + "time" + + "github.com/google/uuid" +) + +// Dispatcher 是 notify 模块对外的发送门面。Repository 由调用方注入(生产 +// 用 pgRepo,测试用 fake),Notifier 通过 Register 动态追加,now 钩子留给 +// 测试覆盖时间相关行为。 +type Dispatcher struct { + repo Repository + notifiers []Notifier + log *slog.Logger + now func() time.Time +} + +// NewDispatcher 用 Repository 与 Logger 构造 Dispatcher。注册 Notifier 通过 +// 后续的 Register 调用完成;调用方应保证在 Dispatch 之前完成所有注册, +// Register 与 Dispatch 不保证并发安全。 +func NewDispatcher(repo Repository, log *slog.Logger) *Dispatcher { + return &Dispatcher{repo: repo, log: log, now: time.Now} +} + +// Register 追加一个 Notifier 到分发列表。 +func (d *Dispatcher) Register(n Notifier) { + d.notifiers = append(d.notifiers, n) +} + +// Dispatch 同步落库;落库失败返回错误,不再异步投递。落库成功后, +// fire-and-forget 派发给所有外部 notifier,单个 notifier 的失败仅记日志, +// 不阻塞其它通道;调用方无法感知异步投递结果。 +func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error { + if msg.ID == uuid.Nil { + msg.ID = uuid.New() + } + if msg.CreatedAt.IsZero() { + msg.CreatedAt = d.now() + } + if err := d.repo.Insert(ctx, msg); err != nil { + return err + } + for _, n := range d.notifiers { + go func(n Notifier) { + // 异步投递使用独立 ctx,避免请求 ctx 取消导致投递中断; + // 真实 notifier 应在内部加超时限制。 + if err := n.Send(context.Background(), msg); err != nil { + d.log.Error("notify channel failed", + "channel", n.Name(), + "topic", msg.Topic, + "err", err.Error(), + ) + } + }(n) + } + return nil +} diff --git a/internal/infra/notify/dispatcher_test.go b/internal/infra/notify/dispatcher_test.go new file mode 100644 index 0000000..f67c81c --- /dev/null +++ b/internal/infra/notify/dispatcher_test.go @@ -0,0 +1,99 @@ +package notify + +import ( + "context" + "errors" + "io" + "log/slog" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +type fakeRepo struct { + mu sync.Mutex + inserted []Message + failNext bool +} + +func (f *fakeRepo) Insert(_ context.Context, msg Message) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.failNext { + f.failNext = false + return errors.New("db down") + } + f.inserted = append(f.inserted, msg) + return nil +} +func (f *fakeRepo) List(context.Context, uuid.UUID, bool, int) ([]Message, error) { return nil, nil } +func (f *fakeRepo) CountUnread(context.Context, uuid.UUID) (int, error) { return 0, nil } +func (f *fakeRepo) MarkRead(context.Context, uuid.UUID, uuid.UUID) error { return nil } +func (f *fakeRepo) MarkAllRead(context.Context, uuid.UUID) error { return nil } + +type spyNotifier struct { + name Channel + count atomic.Int32 + fail bool +} + +func (s *spyNotifier) Name() Channel { return s.name } +func (s *spyNotifier) Send(context.Context, Message) error { + s.count.Add(1) + if s.fail { + return errors.New("send failed") + } + return nil +} + +func newTestLogger() *slog.Logger { + return slog.New(slog.NewJSONHandler(io.Discard, nil)) +} + +func TestDispatch_PersistsThenFansOut(t *testing.T) { + repo := &fakeRepo{} + spy := &spyNotifier{name: ChannelDummy} + + d := NewDispatcher(repo, newTestLogger()) + d.Register(spy) + + require.NoError(t, d.Dispatch(context.Background(), Message{ + UserID: uuid.New(), Topic: "x", Severity: SeverityInfo, + Title: "t", Body: "b", + })) + + require.Len(t, repo.inserted, 1) + require.NotEqual(t, uuid.Nil, repo.inserted[0].ID, "ID 应被 dispatcher 填充") + + require.Eventually(t, func() bool { return spy.count.Load() == 1 }, time.Second, 10*time.Millisecond) +} + +func TestDispatch_PersistFailureReturnsError_DoesNotFanout(t *testing.T) { + repo := &fakeRepo{failNext: true} + spy := &spyNotifier{name: ChannelDummy} + d := NewDispatcher(repo, newTestLogger()) + d.Register(spy) + + err := d.Dispatch(context.Background(), Message{UserID: uuid.New(), Topic: "x"}) + require.Error(t, err) + time.Sleep(50 * time.Millisecond) + require.Equal(t, int32(0), spy.count.Load()) +} + +func TestDispatch_NotifierFailureDoesNotBlockOthers(t *testing.T) { + repo := &fakeRepo{} + bad := &spyNotifier{name: "bad", fail: true} + good := &spyNotifier{name: "good"} + d := NewDispatcher(repo, newTestLogger()) + d.Register(bad) + d.Register(good) + + require.NoError(t, d.Dispatch(context.Background(), Message{UserID: uuid.New()})) + require.Eventually(t, func() bool { + return bad.count.Load() == 1 && good.count.Load() == 1 + }, time.Second, 10*time.Millisecond) +}