feat(server/notices): Store — 广播∪定向合并查询/已读水位/事务内插通知

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01P9G7E3wmAYL9KeYCVZVsqu
2026-07-13 13:36:32 +08:00
parent 542954e4ea
commit ed1f805cbc
2 changed files with 316 additions and 0 deletions
@@ -0,0 +1,220 @@
+// Package notices 承载系统通知(广播 + 定向)的数据访问层:合并查询、已读水位、
+// 事务内插入(供其他领域事件钩子复用)、后台管理(nodectl)增删改查。
+package notices
+
+import (
+	"context"
+	"database/sql"
+	"time"
+)
+
+// Notice 是对外(API/nodectl)的通知视图。json tag 是 API 契约,勿改。
+type Notice struct {
+	ID          int64     `json:"id"`
+	Type        string    `json:"type"`
+	UserID      *int64    `json:"-"`
+	TitleZH     string    `json:"title_zh"`
+	TitleEN     string    `json:"title_en"`
+	BodyZH      string    `json:"body_zh,omitempty"`
+	BodyEN      string    `json:"body_en,omitempty"`
+	Link        string    `json:"link,omitempty"`
+	PublishedAt time.Time `json:"published_at"`
+	Unread      bool      `json:"unread"`
+}
+
+// AdminRow 是 nodectl list 消费的后台视图:不含正文,多一个 status 摘要字段。
+type AdminRow struct {
+	ID          int64
+	Type        string
+	TitleZH     string
+	PublishedAt time.Time
+	Status      string // active|revoked|expired
+}
+
+type Store struct{ db *sql.DB }
+
+func NewStore(db *sql.DB) *Store { return &Store{db: db} }
+
+// ListForUser 返回广播(user_id IS NULL)∪ 定向(user_id=uid)的未撤回、未过期通知,
+// 按 published_at 倒序,并根据 users.notices_read_at 水位计算每条 Unread 与 unreadCount。
+// 水位为 NULL 时按零值时间处理,即全部未读。
+func (s *Store) ListForUser(ctx context.Context, uid int64, now time.Time, limit int) ([]Notice, int, error) {
+	var readAt sql.NullTime
+	if err := s.db.QueryRowContext(ctx, `SELECT notices_read_at FROM users WHERE id=?`, uid).Scan(&readAt); err != nil {
+		return nil, 0, err
+	}
+	watermark := time.Time{}
+	if readAt.Valid {
+		watermark = readAt.Time
+	}
+
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT id, type, user_id, title_zh, title_en, COALESCE(body_zh,''), COALESCE(body_en,''),
+		       COALESCE(link,''), published_at
+		FROM notices
+		WHERE (user_id IS NULL OR user_id = ?)
+		  AND revoked_at IS NULL
+		  AND (expires_at IS NULL OR expires_at > ?)
+		ORDER BY published_at DESC
+		LIMIT ?
+	`, uid, now, limit)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer rows.Close()
+
+	var items []Notice
+	unreadCount := 0
+	for rows.Next() {
+		var n Notice
+		var userID sql.NullInt64
+		if err := rows.Scan(&n.ID, &n.Type, &userID, &n.TitleZH, &n.TitleEN, &n.BodyZH, &n.BodyEN, &n.Link, &n.PublishedAt); err != nil {
+			return nil, 0, err
+		}
+		if userID.Valid {
+			id := userID.Int64
+			n.UserID = &id
+		}
+		n.Unread = n.PublishedAt.After(watermark)
+		if n.Unread {
+			unreadCount++
+		}
+		items = append(items, n)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, 0, err
+	}
+	return items, unreadCount, nil
+}
+
+// MarkRead 把用户的通知已读水位推进到 at。
+func (s *Store) MarkRead(ctx context.Context, uid int64, at time.Time) error {
+	_, err := s.db.ExecContext(ctx, `UPDATE users SET notices_read_at=? WHERE id=?`, at, uid)
+	return err
+}
+
+// InsertNoticeTx 在调用方已开的事务内插入一条定向通知(不自开/自提交事务),
+// 供其他领域的事件钩子(如奖励发放)在同一事务里原子写入。
+func (s *Store) InsertNoticeTx(ctx context.Context, tx *sql.Tx, userID int64, typ, titleZH, titleEN, bodyZH, bodyEN, link string, now time.Time) error {
+	_, err := tx.ExecContext(ctx, `
+		INSERT INTO notices (type, user_id, title_zh, title_en, body_zh, body_en, link, published_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+	`, typ, userID, titleZH, titleEN, nullIfEmpty(bodyZH), nullIfEmpty(bodyEN), nullIfEmpty(link), now)
+	return err
+}
+
+// InsertBroadcast 插入一条全员广播通知(user_id=NULL),自管连接,返回新 id。供
+// nodectl / 发版流程使用。expiresAt 可为 nil(永不过期)。
+func (s *Store) InsertBroadcast(ctx context.Context, typ, titleZH, titleEN, bodyZH, bodyEN, link string, publishedAt time.Time, expiresAt *time.Time) (int64, error) {
+	res, err := s.db.ExecContext(ctx, `
+		INSERT INTO notices (type, user_id, title_zh, title_en, body_zh, body_en, link, published_at, expires_at)
+		VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?)
+	`, typ, titleZH, titleEN, nullIfEmpty(bodyZH), nullIfEmpty(bodyEN), nullIfEmpty(link), publishedAt, nullTime(expiresAt))
+	if err != nil {
+		return 0, err
+	}
+	return res.LastInsertId()
+}
+
+// Revoke 撤回一条通知(置 revoked_at)。目标不存在或已撤回时返回 sql.ErrNoRows。
+func (s *Store) Revoke(ctx context.Context, id int64, at time.Time) error {
+	res, err := s.db.ExecContext(ctx, `UPDATE notices SET revoked_at=? WHERE id=? AND revoked_at IS NULL`, at, id)
+	if err != nil {
+		return err
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if n == 0 {
+		return sql.ErrNoRows
+	}
+	return nil
+}
+
+// ListAdmin 供 nodectl list 使用:全字段倒序;includeInactive=false 时只保留
+// 未撤回、未过期(与 ListForUser 相同判据)的记录。
+func (s *Store) ListAdmin(ctx context.Context, includeInactive bool, limit int) ([]AdminRow, error) {
+	now := time.Now().UTC()
+	query := `
+		SELECT id, type, title_zh, published_at, revoked_at, expires_at
+		FROM notices
+	`
+	args := []any{}
+	if !includeInactive {
+		query += ` WHERE revoked_at IS NULL AND (expires_at IS NULL OR expires_at > ?)`
+		args = append(args, now)
+	}
+	query += ` ORDER BY published_at DESC LIMIT ?`
+	args = append(args, limit)
+
+	rows, err := s.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []AdminRow
+	for rows.Next() {
+		var r AdminRow
+		var revokedAt, expiresAt sql.NullTime
+		if err := rows.Scan(&r.ID, &r.Type, &r.TitleZH, &r.PublishedAt, &revokedAt, &expiresAt); err != nil {
+			return nil, err
+		}
+		switch {
+		case revokedAt.Valid:
+			r.Status = "revoked"
+		case expiresAt.Valid && !expiresAt.Time.After(now):
+			r.Status = "expired"
+		default:
+			r.Status = "active"
+		}
+		out = append(out, r)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// MarkEmailSent 标记该通知的邮件兜底已发送(幂等:只在未发送过时置位)。
+func (s *Store) MarkEmailSent(ctx context.Context, id int64, at time.Time) error {
+	_, err := s.db.ExecContext(ctx, `UPDATE notices SET email_sent_at=? WHERE id=? AND email_sent_at IS NULL`, at, id)
+	return err
+}
+
+// ListActiveUserEmails 返回所有 active 用户的邮箱,供邮件兜底批量发送。
+func (s *Store) ListActiveUserEmails(ctx context.Context) ([]string, error) {
+	rows, err := s.db.QueryContext(ctx, `SELECT email FROM users WHERE status='active'`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []string
+	for rows.Next() {
+		var email string
+		if err := rows.Scan(&email); err != nil {
+			return nil, err
+		}
+		out = append(out, email)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func nullIfEmpty(s string) any {
+	if s == "" {
+		return nil
+	}
+	return s
+}
+
+func nullTime(t *time.Time) any {
+	if t == nil {
+		return nil
+	}
+	return *t
+}
@@ -0,0 +1,96 @@
+package notices
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/config"
+	"github.com/wangjia/pangolin/server/internal/store"
+)
+
+func openDB(t *testing.T) *sql.DB {
+	t.Helper()
+	db, err := store.Open(&config.Config{Driver: "sqlite", DSN: ":memory:"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	if err := store.MigrateUp(db, "sqlite"); err != nil {
+		t.Fatal(err)
+	}
+	_ = store.ApplyCodesLibMigrations(context.Background(), db, "sqlite")
+	return db
+}
+
+func seedU(t *testing.T, db *sql.DB, id int64, uuid string) {
+	t.Helper()
+	if _, err := db.Exec(`INSERT INTO users (id,uuid,email,pw_hash,dp_uuid,status,created_at)
+		VALUES (?,?,?, 'x','dp-'||?, 'active', ?)`, id, uuid, uuid+"@x", uuid, time.Now().UTC()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestListForUser_MergeFilterUnread(t *testing.T) {
+	db := openDB(t)
+	seedU(t, db, 1, "u1")
+	seedU(t, db, 2, "u2")
+	st := NewStore(db)
+	now := time.Now().UTC()
+	// 广播1条 + 我的定向1条 + 他人定向1条 + 已撤回1条 + 已过期1条
+	if _, err := st.InsertBroadcast(context.Background(), "news", "b1", "b1", "", "", "", now.Add(-3*time.Hour), nil); err != nil {
+		t.Fatal(err)
+	}
+	tx, _ := db.Begin()
+	if err := st.InsertNoticeTx(context.Background(), tx, 1, "reward", "mine", "mine", "", "", "", now.Add(-2*time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	if err := st.InsertNoticeTx(context.Background(), tx, 2, "reward", "other", "other", "", "", "", now.Add(-1*time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	_ = tx.Commit()
+	rid, _ := st.InsertBroadcast(context.Background(), "promo", "revoked", "revoked", "", "", "", now, nil)
+	_ = st.Revoke(context.Background(), rid, now)
+	exp := now.Add(-time.Minute)
+	_, _ = st.InsertBroadcast(context.Background(), "promo", "expired", "expired", "", "", "", now.Add(-4*time.Hour), &exp)
+
+	items, unread, err := st.ListForUser(context.Background(), 1, now, 50)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(items) != 2 {
+		t.Fatalf("items=%d want 2(广播+我的;他人/撤回/过期均排除): %+v", len(items), items)
+	}
+	if items[0].TitleZH != "mine" || items[1].TitleZH != "b1" {
+		t.Fatalf("排序应 published_at 倒序: %+v", items)
+	}
+	if unread != 2 {
+		t.Fatalf("水位为空→全未读, unread=%d", unread)
+	}
+
+	// 置水位到 -2.5h:广播(-3h)已读、定向(-2h)未读
+	if err := st.MarkRead(context.Background(), 1, now.Add(-150*time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	items, unread, _ = st.ListForUser(context.Background(), 1, now, 50)
+	if unread != 1 || !items[0].Unread || items[1].Unread {
+		t.Fatalf("水位判定错: unread=%d items=%+v", unread, items)
+	}
+}
+
+func TestInsertNoticeTx_RollsBackWithTx(t *testing.T) {
+	db := openDB(t)
+	seedU(t, db, 1, "u1")
+	st := NewStore(db)
+	tx, _ := db.Begin()
+	if err := st.InsertNoticeTx(context.Background(), tx, 1, "reward", "t", "t", "", "", "", time.Now().UTC()); err != nil {
+		t.Fatal(err)
+	}
+	_ = tx.Rollback()
+	var n int
+	_ = db.QueryRow(`SELECT COUNT(*) FROM notices`).Scan(&n)
+	if n != 0 {
+		t.Fatalf("回滚后应零孤儿通知, got %d", n)
+	}
+}