fix(notify): waitgroup-based dispatcher shutdown, withoutcancel persist

internal/app/app.go

@@ -42,6 +42,7 @@ type App struct {
 	log        *slog.Logger
 	pool       *db.Pool
 	server     *http.Server
+	dispatcher *notify.Dispatcher
 }
 
 // New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误，不留半成品资源。
@@ -68,9 +69,6 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
 	notifyRepo := notify.NewPostgresRepository(pool)
 	notifyDispatcher := notify.NewDispatcher(notifyRepo, log)
 	notifyDispatcher.Register(notify.NewDummyNotifier(log))
-	// 当前 dispatcher 仅暴露给本包：未来引入定时任务/后台 worker 才需把它
-	// 通过 App 字段透出去，目前保留局部变量以避免暴露未稳定 API。
-	_ = notifyDispatcher
 
 	spaSub, err := fs.Sub(dist, "web/dist")
 	if err != nil {
@@ -113,7 +111,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
 		WriteTimeout:      60 * time.Second,
 		IdleTimeout:       120 * time.Second,
 	}
-	return &App{cfg: cfg, log: log, pool: pool, server: srv}, nil
+	return &App{cfg: cfg, log: log, pool: pool, server: srv, dispatcher: notifyDispatcher}, nil
 }
 
 // Run 启动 HTTP server 并阻塞直到 ctx 取消（Shutdown）或服务器报错。
@@ -137,9 +135,16 @@ func (a *App) Run(ctx context.Context) error {
 		if shutErr != nil {
 			a.log.Error("server shutdown", "err", shutErr.Error())
 		}
+		// server 已停止接收新请求；等异步通知投递结束再关 pool
+		if err := a.dispatcher.Shutdown(shutCtx); err != nil {
+			a.log.Warn("notify dispatcher drain", "err", err.Error())
+		}
 		a.pool.Close()
 		return shutErr
 	case err := <-errCh:
+		drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = a.dispatcher.Shutdown(drainCtx)
 		a.pool.Close()
 		return err
 	}

internal/infra/notify/dispatcher.go

@@ -6,6 +6,7 @@ package notify
 import (
 	"context"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -13,12 +14,13 @@ import (
 
 // Dispatcher 是 notify 模块对外的发送门面。Repository 由调用方注入（生产
 // 用 pgRepo，测试用 fake），Notifier 通过 Register 动态追加，now 钩子留给
-// 测试覆盖时间相关行为。
+// 测试覆盖时间相关行为。wg 用于 Shutdown 时等待所有异步投递结束。
 type Dispatcher struct {
 	repo      Repository
 	notifiers []Notifier
 	log       *slog.Logger
 	now       func() time.Time
+	wg        sync.WaitGroup
 }
 
 // NewDispatcher 用 Repository 与 Logger 构造 Dispatcher。注册 Notifier 通过
@@ -36,6 +38,10 @@ func (d *Dispatcher) Register(n Notifier) {
 // Dispatch 同步落库；落库失败返回错误，不再异步投递。落库成功后，
 // fire-and-forget 派发给所有外部 notifier，单个 notifier 的失败仅记日志，
 // 不阻塞其它通道；调用方无法感知异步投递结果。
+//
+// 落库使用 context.WithoutCancel + 5s 硬超时，避免请求 ctx 取消导致通知
+// 漏写；外部投递使用独立 background ctx + 30s 硬超时，lifecycle 由 Shutdown
+// 通过 wg 管控。
 func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error {
 	if msg.ID == uuid.Nil {
 		msg.ID = uuid.New()
@@ -43,14 +49,18 @@ func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error {
 	if msg.CreatedAt.IsZero() {
 		msg.CreatedAt = d.now()
 	}
-	if err := d.repo.Insert(ctx, msg); err != nil {
+	persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+	defer cancel()
+	if err := d.repo.Insert(persistCtx, msg); err != nil {
 		return err
 	}
 	for _, n := range d.notifiers {
+		d.wg.Add(1)
 		go func(n Notifier) {
-			// 异步投递使用独立 ctx，避免请求 ctx 取消导致投递中断；
+			defer d.wg.Done()
-			// 真实 notifier 应在内部加超时限制。
+			sendCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-			if err := n.Send(context.Background(), msg); err != nil {
+			defer cancel()
+			if err := n.Send(sendCtx, msg); err != nil {
 				d.log.Error("notify channel failed",
 					"channel", n.Name(),
 					"topic", msg.Topic,
@@ -61,3 +71,20 @@ func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error {
 	}
 	return nil
 }
+
+// Shutdown 阻塞等待所有异步投递完成。调用方应保证此后不再有新的 Dispatch
+// 调用（通常上游 HTTP server 已经 Shutdown）。ctx 用作硬截止：超时会返回
+// ctx.Err()，但泄漏的 goroutine 会继续跑到 Send 的 30s timeout 自然结束。
+func (d *Dispatcher) Shutdown(ctx context.Context) error {
+	done := make(chan struct{})
+	go func() {
+		d.wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}