Documentation
¶
Overview ¶
Package notify 多实例变更广播（#14 T1 基础层）：PG LISTEN/NOTIFY 定向刷新。
架构（设计文档 docs/superpowers/plans/2026-08-10-multi-instance-design.md §2）：管理面变更落库成功后经 Publisher 发一条 NOTIFY（单 channel c3api_invalidate，紧凑 JSON 载荷与 invalidate.State 同构）；每实例一个 Listener worker（Name="notify"）LISTEN 该 channel，解析后调注入的 Dispatcher（main 装配，T3）转发现有 invalidate.Debouncer 的 Mark 方法—— 本地/远端变更共享同一去抖窗口，天然合并去重，Debouncer 本体零改动。
设计要点：
- 载荷守卫：PG NOTIFY 载荷上限 8000B，批量账号变更的 Groups 集合可能 超限 → marshal 后 > 6KB 时丢弃 Groups 并置 Templates=true（降级 sched 全量重载，sched 全量包含组级重载，语义仍正确）。
- Src 自播跳过：Publisher 自动填 src=实例 ID，接收端跳过自身发布的 NOTIFY（省一次重复 reload）。
- 计费扣费路径绝不发布 NOTIFY（每 flush 即风暴），保持现状语义。
Index ¶
Constants ¶
const Channel = "c3api_invalidate"
Channel NOTIFY 频道名（单 channel：与去抖器合并 State 对齐——一类资源一个 channel 则监听器多、且与 debouncer 的合并 State 不对齐）。
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Change ¶
type Change struct {
// V 载荷版本（当前 1）；未来字段演进/接收端兼容判读用。
V int `json:"v"`
// Users KindUsers：用户 CRUD（含创建）/余额变更 → auth + 余额快照全量。
Users bool `json:"users,omitempty"`
// Templates KindTemplates：模板（base_url/models/映射）变更 → sched 全量
// + clients 失效。载荷守卫降级 full 时也置此位。
Templates bool `json:"templates,omitempty"`
// Clients KindClients：aiclient 工厂失效（模板 base_url / 账号
// upstream_key 变更）。
Clients bool `json:"clients,omitempty"`
// Multipliers KindMultipliers：组倍率 / 用户-组专属倍率变更 → 余额倍率
// 快照定向刷新。
Multipliers bool `json:"multipliers,omitempty"`
// Keys key CRUD 缺口（创建/轮换/删除/改额度）→ auth 快照全量 Reload
//（v1 不做增量定向）。
Keys bool `json:"keys,omitempty"`
// Settings settings 快照变更（UpdateSetting）→ settings 快照重载。
Settings bool `json:"settings,omitempty"`
// Rules 规则表变更（规则 CRUD）→ 规则表重载（重载清窗口计数，全实例
// 同步执行语义）。
Rules bool `json:"rules,omitempty"`
// Groups 组级定向（账号变更的受影响组 id）。
Groups []int64 `json:"groups,omitempty"`
// Src 发布实例 ID：接收端跳过自播（省一次重复 reload）。Publisher 自动
// 填，调用方无需关心。
Src string `json:"src,omitempty"`
}
Change 一条 NOTIFY 变更载荷（与 invalidate.State 同构，紧凑 JSON： omitempty + 短字段名，尽量压载荷）。
type Conn ¶
type Conn interface {
// Listen 注册 LISTEN channel。
Listen(ctx context.Context, channel string) error
// WaitForNotification 阻塞等待一条通知；连接断开/ctx 取消返回错误。
WaitForNotification(ctx context.Context) (*pgconn.Notification, error)
// Close 关闭连接（幂等）。
Close(ctx context.Context) error
}
Conn 监听连接面：生产实现包装独立 *pgx.Conn，测试注入 fake。
type ConnectFunc ¶
ConnectFunc 建立监听连接（默认 pgx.Connect；测试注入假连接）。
type Dispatcher ¶
type Dispatcher interface {
// Apply 处理一条 NOTIFY 变更：转发给 invalidate.Debouncer 的 Mark 方法
//（本地/远端合并同窗口，天然去重）——users/templates/multipliers/groups
// 转现有 Mark（Templates 含 clients 失效）；keys/rules 转 Keys/Rules 分支
// （reloadAll 扩展）；settings 由装配侧同步 ReloadSettings + 注册表 scope
// 分发（#36 时序——scope 重载必须读到新 N，实现见 cmd/server dispatcher）。
// 无返回值：内部失败由实现独立 Warn 消化（G-P2-1——NOTIFY 是事件提示，
// 调用方无任何可执行动作，透传只会双 Warn；周期 ticker / 60s 兜底已存在）。
Apply(ctx context.Context, ch Change)
// FullRefresh 连接成功（启动首连 / 断线重连）时的本地刷新：Auth Reload +
// Balances Reload + sched InvalidateAll + settings + rules 重载（覆盖
// 断连期间 NOTIFY 丢失；设计文档 §2.3 / R1 / R8）。E2：首连且启动首刷全
// 成功时实现可跳过全量仅补 settings（装配侧 dispatcher 注释）。
FullRefresh(ctx context.Context) error
}
Dispatcher 变更分发回调（main 装配实现；避免 notify → invalidate/service 依赖环——notify 只依赖本接口，invalidate 不 import notify）。
实现必须尊重 ctx 取消（Apply/FullRefresh 内部的长操作须响应 ctx.Done）： Close 会等待监听 goroutine 退出，不响应取消的调用会阻塞停机。
type Execer ¶
type Execer interface {
Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
}
Execer 发布端依赖的最小执行面：*pgxpool.Pool（生产，repository.OpenPG 的 池）与 pgxmock.PgxPoolIface（测试）均满足。
type Listener ¶
type Listener struct {
// contains filtered or unexported fields
}
Listener NOTIFY 监听 worker（worker.Worker 契约，Name="notify"）：
- 独立单连接 LISTEN c3api_invalidate（重连理由见 pgxConnect 注释）；
- 循环消费通知：解析 Change → 自播（Src == 本实例）跳过 → Dispatcher.Apply；
- 连接断开 → 指数退避重连（1s→30s cap）；连接成功（启动首连或重连）立即 执行一次 Dispatcher.FullRefresh（覆盖断连期间 NOTIFY 丢失，R8；E2：首连 且启动首刷全成功 → dispatcher 跳过五路仅补 settings；60s 周期兜底是另 一层，见设计文档 §5 #9）；
- Close：取消循环 + 等 goroutine 退出（幂等；未 Start 也安全）。
func (*Listener) Close ¶
Close 停止监听（幂等；未 Start 也安全）：取消循环 → 等 goroutine 退出。 循环的阻塞点（WaitForNotification / 退避 sleep）都响应 ctx 取消，立即返回。
func (*Listener) Start ¶
Start 启动监听 goroutine（幂等：重复 Start 返回错误；worker 契约）。非阻塞： 首连/全量刷新在 goroutine 内异步完成。
cancel/done 在 spawn 之前赋值——Close 与 Start 并发时必能看到句柄（不存在 "goroutine 已启动但 Close 读到 nil cancel"的窗口；Manager 串行化 Start/Close， 先于未完成 Start 的 Close 由 Start ctx 取消兜底）。失败路径（重复 Start）经 defer 释放本次创建的 ctx，不泄漏。
type ListenerConfig ¶
type ListenerConfig struct {
DSN string // 独立监听连接 DSN（与业务池同 DSN）
Src string // 实例 ID：跳过自播 NOTIFY（空 = 不跳过）
Channel string // 空 → Channel（c3api_invalidate）
Dispatcher Dispatcher // 必填（nil → Start 返回错误）
Log *logx.Logger // 可空（nil = 不记日志）
Connect ConnectFunc // nil → 默认 pgx 独立连接
// BackoffBase/BackoffMax 断线重连退避：0 → 1s / 30s。
BackoffBase time.Duration
BackoffMax time.Duration
}
ListenerConfig 监听器装配参数。
type ListenerStats ¶
type ListenerStats struct {
Running bool `json:"running"` // 监听循环存活（Start 置位、run 退出复位；原子读零锁）
}
ListenerStats NOTIFY 监听 worker 状态（存活最小集）。
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher NOTIFY 发布器：marshal → SELECT pg_notify('c3api_invalidate', $1)。 独立持有 pool 引用（构造注入，走现有 repository.OpenPG 的 pgxpool），不 新建连接。
func NewPublisher ¶
NewPublisher 构造发布器。pool 必填（未注入时 Publish 返回显式错误——装配 缺失显式暴露，不静默 no-op）。
Source Files