You've already forked agentic-coding-workflow
102 lines
3.4 KiB
Go
102 lines
3.4 KiB
Go
// Package notify 内的 dispatcher.go 实现通知分发的核心：先同步落库，再
|
|
// 异步把消息广播到所有已注册的 Notifier。落库失败立即返回错误，外部投递
|
|
// 不会启动；外部投递失败仅记日志，不影响调用方与其它通道。
|
|
package notify
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// 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
|
|
// hub 是可选的实时推送中心；落库成功后 Publish。nil 时仅持久化 + 外发通道。
|
|
hub StreamHub
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
// SetStreamHub 注入实时推送中心（装配期）。落库成功后会向其 Publish。
|
|
func (d *Dispatcher) SetStreamHub(h StreamHub) {
|
|
d.hub = h
|
|
}
|
|
|
|
// 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()
|
|
}
|
|
if msg.CreatedAt.IsZero() {
|
|
msg.CreatedAt = d.now()
|
|
}
|
|
persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
if err := d.repo.Insert(persistCtx, msg); err != nil {
|
|
return err
|
|
}
|
|
// 落库成功后才推送实时事件（推送失败 / 慢订阅者不影响落库的权威 inbox）。
|
|
if d.hub != nil {
|
|
d.hub.Publish(msg.UserID, msg)
|
|
}
|
|
for _, n := range d.notifiers {
|
|
d.wg.Add(1)
|
|
go func(n Notifier) {
|
|
defer d.wg.Done()
|
|
sendCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
if err := n.Send(sendCtx, msg); err != nil {
|
|
d.log.Error("notify channel failed",
|
|
"channel", n.Name(),
|
|
"topic", msg.Topic,
|
|
"err", err.Error(),
|
|
)
|
|
}
|
|
}(n)
|
|
}
|
|
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()
|
|
}
|
|
}
|