diff --git a/internal/infra/notify/notify.go b/internal/infra/notify/notify.go index 7e23b7c..240ba02 100644 --- a/internal/infra/notify/notify.go +++ b/internal/infra/notify/notify.go @@ -53,6 +53,8 @@ type Message struct { Body string Link string Metadata map[string]any + // ReadAt 表示该通知的已读时间。nil 表示未读,非 nil 表示用户已读的时刻。 + ReadAt *time.Time CreatedAt time.Time } diff --git a/internal/infra/notify/repository.go b/internal/infra/notify/repository.go new file mode 100644 index 0000000..2a20e73 --- /dev/null +++ b/internal/infra/notify/repository.go @@ -0,0 +1,199 @@ +// Package notify 内的 repository.go 是 notifysqlc 生成查询包之上的薄包装层。 +// 职责:将 "驱动层类型"(pgtype.UUID / pgtype.Timestamptz)与 "领域类型" +// (uuid.UUID / time.Time / *time.Time)做双向转换,并把底层驱动错误统一包装 +// 为 errs.AppError,使 service 层可以面向干净的 Repository 接口编程,无需直接 +// 依赖 pgx 驱动细节。 +// +// 设计取舍: +// - Repository interface 提供可替换的注入点,供 F4/F5 阶段 fake 实现注入。 +// - pgtype 转换适配器刻意保持包内私有,避免对外泄露驱动依赖;若其它子包 +// 需要类似映射,应各自维护或上抽到 infra/db。 +// - Insert 中 Metadata 字段用 json.Marshal 序列化;空 map 与 nil 均会写入 +// 合法 JSON,保证列约束不违反。 +package notify + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/yan1h/agent-coding-workflow/internal/infra/errs" + notifysqlc "github.com/yan1h/agent-coding-workflow/internal/infra/notify/sqlc" +) + +// Repository 抽象 notify 模块对持久层的全部依赖。所有方法接受 context 并 +// 返回领域类型,既方便 service 层 mock,也方便日后切换驱动实现。 +type Repository interface { + Insert(ctx context.Context, msg Message) error + List(ctx context.Context, userID uuid.UUID, unreadOnly bool, limit int) ([]Message, error) + CountUnread(ctx context.Context, userID uuid.UUID) (int, error) + MarkRead(ctx context.Context, userID, msgID uuid.UUID) error + MarkAllRead(ctx context.Context, userID uuid.UUID) error +} + +// pgRepo 是 Repository 的 Postgres 实现,内部持有 sqlc 生成的 Queries。 +// 字段保持小写避免外泄;外部只能通过 NewPostgresRepository 构造。 +type pgRepo struct { + q *notifysqlc.Queries +} + +// NewPostgresRepository 用现有的 pgxpool 构造 Repository 实现。pool 必须由 +// 调用方负责 Close;本仓库不接管其生命周期。 +func NewPostgresRepository(pool *pgxpool.Pool) Repository { + return &pgRepo{q: notifysqlc.New(pool)} +} + +// Insert 把一条 Message 持久化到 notifications 表。Metadata 先经 json.Marshal +// 序列化为字节,Link 空字符串写为 NULL(见 ptr 函数语义)。 +func (r *pgRepo) Insert(ctx context.Context, msg Message) error { + meta, err := json.Marshal(msg.Metadata) + if err != nil { + return errs.Wrap(err, errs.CodeInternal, "marshal metadata") + } + if err := r.q.InsertNotification(ctx, notifysqlc.InsertNotificationParams{ + ID: toPgUUID(msg.ID), + UserID: toPgUUID(msg.UserID), + Topic: msg.Topic, + Severity: string(msg.Severity), + Title: msg.Title, + Body: msg.Body, + Link: ptr(msg.Link), + Metadata: meta, + CreatedAt: toPgTimestamptz(msg.CreatedAt), + }); err != nil { + return errs.Wrap(err, errs.CodeInternal, "insert notification") + } + return nil +} + +// List 查询指定用户的通知列表。unreadOnly=true 时仅返回未读条目;limit 控制 +// 最大返回行数,由调用方保证为合理正整数。 +func (r *pgRepo) List(ctx context.Context, userID uuid.UUID, unreadOnly bool, limit int) ([]Message, error) { + rows, err := r.q.ListNotificationsByUser(ctx, notifysqlc.ListNotificationsByUserParams{ + UserID: toPgUUID(userID), + Column2: unreadOnly, + Limit: int32(limit), + }) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "list notifications") + } + + msgs := make([]Message, 0, len(rows)) + for _, row := range rows { + var meta map[string]any + if len(row.Metadata) > 0 { + if err := json.Unmarshal(row.Metadata, &meta); err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "unmarshal notification metadata") + } + } + msgs = append(msgs, Message{ + ID: fromPgUUID(row.ID), + UserID: fromPgUUID(row.UserID), + Topic: row.Topic, + Severity: Severity(row.Severity), + Title: row.Title, + Body: row.Body, + Link: deref(row.Link), + Metadata: meta, + ReadAt: fromPgTimestamptzPtr(row.ReadAt), + CreatedAt: fromPgTimestamptz(row.CreatedAt), + }) + } + return msgs, nil +} + +// CountUnread 返回指定用户的未读通知数量。 +func (r *pgRepo) CountUnread(ctx context.Context, userID uuid.UUID) (int, error) { + n, err := r.q.CountUnreadByUser(ctx, toPgUUID(userID)) + if err != nil { + return 0, errs.Wrap(err, errs.CodeInternal, "count unread") + } + return int(n), nil +} + +// MarkRead 将单条通知标记为已读。msgID 与 userID 同时匹配才会更新,防止 +// 用户越权操作他人通知(数据库层已作 WHERE user_id = $2 约束)。 +func (r *pgRepo) MarkRead(ctx context.Context, userID, msgID uuid.UUID) error { + if err := r.q.MarkRead(ctx, notifysqlc.MarkReadParams{ + ID: toPgUUID(msgID), + UserID: toPgUUID(userID), + }); err != nil { + return errs.Wrap(err, errs.CodeInternal, "mark read") + } + return nil +} + +// MarkAllRead 将指定用户的所有未读通知一次性标记为已读。 +func (r *pgRepo) MarkAllRead(ctx context.Context, userID uuid.UUID) error { + if err := r.q.MarkAllReadByUser(ctx, toPgUUID(userID)); err != nil { + return errs.Wrap(err, errs.CodeInternal, "mark all read") + } + return nil +} + +// -------------------------------------------------------------------------- +// 包内私有适配器:pgtype ↔ 领域类型转换 +// -------------------------------------------------------------------------- + +// toPgUUID 把领域 uuid.UUID 转为 sqlc 期望的 pgtype.UUID。 +// 始终标记 Valid=true;调用方传 uuid.Nil 即视为显式写入零值。 +func toPgUUID(u uuid.UUID) pgtype.UUID { + return pgtype.UUID{Bytes: u, Valid: true} +} + +// fromPgUUID 把 pgtype.UUID 转回领域 uuid.UUID。Valid=false 时返回 uuid.Nil, +// 让上层用零值判定 "无值",无需关心驱动层的可空封装。 +func fromPgUUID(p pgtype.UUID) uuid.UUID { + if !p.Valid { + return uuid.Nil + } + return uuid.UUID(p.Bytes) +} + +// toPgTimestamptz 把 time.Time 转为 pgtype.Timestamptz,始终标 Valid=true。 +// IsZero 的时间由 dispatcher 或上层兜底,repository 层不做额外校验。 +func toPgTimestamptz(t time.Time) pgtype.Timestamptz { + return pgtype.Timestamptz{Time: t, Valid: true} +} + +// fromPgTimestamptz 把 pgtype.Timestamptz 转回 time.Time。Valid=false 时 +// 返回 time.Time 零值,调用方可通过 IsZero() 判断是否有实际时间。 +func fromPgTimestamptz(p pgtype.Timestamptz) time.Time { + if !p.Valid { + return time.Time{} + } + return p.Time +} + +// fromPgTimestamptzPtr 把可空的 pgtype.Timestamptz 转为 *time.Time。 +// Valid=false(即数据库 NULL)返回 nil,表示 "未发生"(例如通知未读时 read_at 为 NULL); +// Valid=true 返回指向时间副本的指针,避免调用方误操作原始值。 +func fromPgTimestamptzPtr(p pgtype.Timestamptz) *time.Time { + if !p.Valid { + return nil + } + t := p.Time + return &t +} + +// ptr 将非空字符串转为指针,空字符串返回 nil,令数据库列写入 NULL。 +// 这与 notifications.link 列的语义一致:无链接时存 NULL 而非空字符串。 +func ptr(s string) *string { + if s == "" { + return nil + } + return &s +} + +// deref 将字符串指针解引用为字符串值,nil 时返回空字符串。 +// 用于把数据库 NULL link 列还原为领域 Message.Link 零值。 +func deref(s *string) string { + if s == nil { + return "" + } + return *s +}