feat(notify): dispatcher with sync persist and async channel fanout

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
+}

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)
+}