feat(notify): postgres repository for notifications inbox

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
 }
 

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