package app import ( "context" "fmt" "net/http" "strconv" "strings" "time" ) // Announcements. Until now operations had no way to say anything to users: // an outage, a rule change, a campaign — all of it depended on people noticing // by themselves. Sending is deliberately recorded and revocable, because the // two questions afterwards are always "who got it" and "can we take it back". type broadcastRequest struct { Title string `json:"title"` Content string `json:"content"` Audience string `json:"audience"` UserIDs []int64 `json:"userIds"` CityCode string `json:"cityCode"` WithPush bool `json:"withPush"` } // broadcastAudience turns the chosen audience into a recipient query. Keeping // the SQL here (rather than building it from request fields) is what stops an // audience picker from turning into an injection point. func (a *App) broadcastAudience(req broadcastRequest) (string, []any, error) { switch strings.TrimSpace(req.Audience) { case "all": return `SELECT id FROM users WHERE status=1 AND deleted_at IS NULL`, nil, nil case "vip": return `SELECT u.id FROM users u JOIN user_profiles p ON p.user_id=u.id WHERE u.status=1 AND u.deleted_at IS NULL AND p.is_vip=1`, nil, nil case "city": code := strings.TrimSpace(req.CityCode) if code == "" { return "", nil, fmt.Errorf("请选择城市") } return `SELECT u.id FROM users u JOIN user_location_states l ON l.user_id=u.id WHERE u.status=1 AND u.deleted_at IS NULL AND l.city_code=?`, []any{code}, nil case "sitters": return `SELECT u.id FROM users u JOIN pet_sitters s ON s.user_id=u.id WHERE u.status=1 AND u.deleted_at IS NULL AND s.status=1`, nil, nil case "owners": return `SELECT DISTINCT u.id FROM users u JOIN pets p ON p.owner_user_id=u.id WHERE u.status=1 AND u.deleted_at IS NULL AND p.deleted_at IS NULL`, nil, nil case "users": if len(req.UserIDs) == 0 { return "", nil, fmt.Errorf("请指定接收的用户") } if len(req.UserIDs) > 500 { return "", nil, fmt.Errorf("指定用户最多 500 个,再多请改用人群条件") } placeholders := make([]string, 0, len(req.UserIDs)) args := make([]any, 0, len(req.UserIDs)) for _, id := range req.UserIDs { placeholders = append(placeholders, "?") args = append(args, id) } return `SELECT id FROM users WHERE status=1 AND deleted_at IS NULL AND id IN (` + strings.Join(placeholders, ",") + `)`, args, nil } return "", nil, fmt.Errorf("人群条件无效") } func (a *App) adminSendBroadcast(w http.ResponseWriter, r *http.Request) { var req broadcastRequest if decode(r, &req) != nil { fail(w, http.StatusBadRequest, 20001, "格式错误") return } req.Title = strings.TrimSpace(req.Title) req.Content = strings.TrimSpace(req.Content) if req.Title == "" || len([]rune(req.Title)) > 40 { fail(w, http.StatusBadRequest, 20001, "标题为 1 到 40 个字") return } if len([]rune(req.Content)) < 5 || len([]rune(req.Content)) > 500 { fail(w, http.StatusBadRequest, 20001, "正文为 5 到 500 个字") return } query, args, err := a.broadcastAudience(req) if err != nil { fail(w, http.StatusBadRequest, 20001, err.Error()) return } rows, err := a.db.QueryContext(r.Context(), query, args...) if err != nil { fail(w, http.StatusInternalServerError, 50001, "查询接收人失败") return } recipients := []int64{} for rows.Next() { var id int64 if rows.Scan(&id) == nil { recipients = append(recipients, id) } } rows.Close() if len(recipients) == 0 { fail(w, http.StatusBadRequest, 20001, "这个条件下没有匹配到用户") return } limit := a.configInt(r.Context(), "notification.broadcast_max_recipients", 20000) if len(recipients) > limit { fail(w, http.StatusBadRequest, 20001, fmt.Sprintf("匹配到 %d 人,超过单次上限 %d 人", len(recipients), limit)) return } reference := strings.TrimSpace(req.CityCode) if req.Audience == "users" { ids := make([]string, 0, len(req.UserIDs)) for _, id := range req.UserIDs { ids = append(ids, strconv.FormatInt(id, 10)) } reference = strings.Join(ids, ",") if len(reference) > 500 { reference = reference[:500] } } withPush := req.WithPush && a.configBool(r.Context(), "notification.broadcast_push_enabled", true) result, err := a.db.ExecContext(r.Context(), `INSERT INTO notification_broadcasts (title,content,audience,audience_ref,with_push,recipient_count,created_by) VALUES(?,?,?,?,?,?,?)`, req.Title, req.Content, req.Audience, reference, withPush, len(recipients), current(r).ID) if err != nil { fail(w, http.StatusInternalServerError, 50001, "创建公告失败") return } broadcastID, _ := result.LastInsertId() // 一条一条插会把几万人的公告拖成分钟级,所以按批写; // 关掉了系统通知的人不发,这是他自己的设置。 delivered := 0 const batchSize = 500 for start := 0; start < len(recipients); start += batchSize { end := start + batchSize if end > len(recipients) { end = len(recipients) } values := make([]string, 0, end-start) params := make([]any, 0, (end-start)*6) for _, userID := range recipients[start:end] { if !a.notificationAllowed(r.Context(), userID, "system") { continue } values = append(values, "(?,?,?,?,?,?)") params = append(params, userID, "system", req.Title, req.Content, "broadcast", broadcastID) } if len(values) == 0 { continue } if _, err = a.db.ExecContext(r.Context(), `INSERT INTO notifications(user_id,type,title,content,biz_type,biz_id) VALUES`+ strings.Join(values, ","), params...); err != nil { fail(w, http.StatusInternalServerError, 50001, "写入通知失败") return } delivered += len(values) } _, _ = a.db.ExecContext(r.Context(), `UPDATE notification_broadcasts SET recipient_count=? WHERE id=?`, delivered, broadcastID) if withPush { // 推送是尽力而为:走不通不该让公告本身失败,所以放到后台并单独计数。 go a.pushBroadcast(context.Background(), broadcastID, recipients, req.Title, req.Content) } a.audit(r, "broadcast", "notification", broadcastID, map[string]any{ "audience": req.Audience, "recipients": delivered, "withPush": withPush, "title": req.Title, }) reply(w, map[string]any{"id": broadcastID, "recipientCount": delivered, "withPush": withPush}) } func (a *App) pushBroadcast(ctx context.Context, broadcastID int64, recipients []int64, title, content string) { credentials, ok := a.pushSettings(ctx) if !ok { return } token, err := a.pushToken(ctx, credentials) if err != nil { return } sent := 0 for _, userID := range recipients { targets := a.pushRecipients(ctx, 0, 0, []int64{userID}) for _, target := range targets { if a.sendPush(ctx, credentials, token, target, title, content, 0) == nil { sent++ } } } _, _ = a.db.ExecContext(ctx, `UPDATE notification_broadcasts SET push_count=? WHERE id=?`, sent, broadcastID) } func (a *App) adminBroadcasts(w http.ResponseWriter, r *http.Request) { page, size, offset := pagination(r) var total int64 _ = a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM notification_broadcasts`).Scan(&total) rows, err := a.db.QueryContext(r.Context(), `SELECT b.id,b.title,b.content,b.audience,b.audience_ref,b.with_push, b.recipient_count,b.push_count,b.status,b.revoked_count,b.created_at,COALESCE(admin.username,''), (SELECT COUNT(*) FROM notifications n WHERE n.biz_type='broadcast' AND n.biz_id=b.id AND n.read_at IS NOT NULL) FROM notification_broadcasts b LEFT JOIN admin_users admin ON admin.id=b.created_by ORDER BY b.id DESC LIMIT ? OFFSET ?`, size, offset) if err != nil { fail(w, http.StatusInternalServerError, 50001, "查询公告失败") return } defer rows.Close() items := []map[string]any{} for rows.Next() { var id int64 var title, content, audience, reference, status, operator string var withPush bool var recipients, pushes, revoked, read int var created time.Time if rows.Scan(&id, &title, &content, &audience, &reference, &withPush, &recipients, &pushes, &status, &revoked, &created, &operator, &read) != nil { continue } items = append(items, map[string]any{ "id": id, "title": title, "content": content, "audience": audience, "audienceRef": reference, "withPush": withPush, "recipientCount": recipients, "pushCount": pushes, "status": status, "revokedCount": revoked, "readCount": read, "operator": operator, "createdAt": created.Format(time.RFC3339), }) } reply(w, map[string]any{"items": items, "total": total, "page": page, "size": size}) } // adminRevokeBroadcast pulls back only what nobody has read yet. A notification // someone already opened cannot be unsent, and pretending otherwise would make // the counts lie. func (a *App) adminRevokeBroadcast(w http.ResponseWriter, r *http.Request) { id, err := pathID(r) if err != nil { fail(w, http.StatusBadRequest, 20001, "编号无效") return } var status string if a.db.QueryRowContext(r.Context(), `SELECT status FROM notification_broadcasts WHERE id=?`, id).Scan(&status) != nil { fail(w, http.StatusNotFound, 30001, "公告不存在") return } if status == "REVOKED" { fail(w, http.StatusBadRequest, 20001, "这条公告已经撤回过了") return } result, err := a.db.ExecContext(r.Context(), `DELETE FROM notifications WHERE biz_type='broadcast' AND biz_id=? AND read_at IS NULL`, id) if err != nil { fail(w, http.StatusInternalServerError, 50001, "撤回失败") return } revoked, _ := result.RowsAffected() if _, err = a.db.ExecContext(r.Context(), `UPDATE notification_broadcasts SET status='REVOKED',revoked_count=?,revoked_at=NOW(3) WHERE id=?`, revoked, id); err != nil { fail(w, http.StatusInternalServerError, 50001, "撤回失败") return } a.audit(r, "revoke_broadcast", "notification", id, map[string]any{"revoked": revoked}) reply(w, map[string]any{"success": true, "revokedCount": revoked}) } // adminBroadcastPreview answers "how many people would get this" before anyone // presses send. func (a *App) adminBroadcastPreview(w http.ResponseWriter, r *http.Request) { req := broadcastRequest{ Audience: strings.TrimSpace(r.URL.Query().Get("audience")), CityCode: strings.TrimSpace(r.URL.Query().Get("cityCode")), } if raw := strings.TrimSpace(r.URL.Query().Get("userIds")); raw != "" { for _, part := range strings.Split(raw, ",") { if id, convErr := strconv.ParseInt(strings.TrimSpace(part), 10, 64); convErr == nil { req.UserIDs = append(req.UserIDs, id) } } } query, args, err := a.broadcastAudience(req) if err != nil { fail(w, http.StatusBadRequest, 20001, err.Error()) return } var count int if a.db.QueryRowContext(r.Context(), `SELECT COUNT(*) FROM (`+query+`) AS audience`, args...).Scan(&count) != nil { fail(w, http.StatusInternalServerError, 50001, "预览失败") return } reply(w, map[string]any{"count": count, "limit": a.configInt(r.Context(), "notification.broadcast_max_recipients", 20000)}) }