252 lines
8.4 KiB
Go
252 lines
8.4 KiB
Go
// Copyright 2026 tamakyi. All rights reserved.
|
||
// Use of this source code is governed by a MIT-style
|
||
// license that can be found in the LICENSE file.
|
||
|
||
package vote
|
||
|
||
import (
|
||
stdctx "context"
|
||
"encoding/json"
|
||
"strconv"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/pkg/errors"
|
||
"github.com/sirupsen/logrus"
|
||
|
||
"github.com/tamakyi/TamaBox/internal/context"
|
||
"github.com/tamakyi/TamaBox/internal/db"
|
||
"github.com/tamakyi/TamaBox/internal/shared"
|
||
)
|
||
|
||
// resultsRefreshing 记录正在后台重算的投票 ID。值为 struct{}，只用于占位去重。
|
||
//
|
||
// 用 sync.Map 而不是「mutex + map」：这是典型的多数读、少数写，且 TryLoadOrStore
|
||
// 一步完成「判重 + 占位」，不需要先查后写那种两步操作（那正是别处幂等缺陷的成因）。
|
||
var resultsRefreshing sync.Map
|
||
|
||
// resultsRefreshTimeout 单次后台重算的上限。
|
||
// 后台任务与请求生命周期解绑，但必须有自己的上限，否则一次全量聚合可能无限跑下去。
|
||
const resultsRefreshTimeout = 60 * time.Second
|
||
|
||
// LoadResults 返回投票的聚合统计，策略是 stale-while-revalidate。
|
||
//
|
||
// 改这里的原因：多选 / 评分 / 排序 / 表单的统计必须在 Go 层逐条 json.Unmarshal
|
||
// 全部提交记录（O(N)，`ListByVoteID(…, nil)` 不设 LIMIT）。而 Submit 每来一票就
|
||
// InvalidateResultsCache，于是持续有人投票时：第 k 票触发一次 O(k) 重算，稳态是 O(N²)。
|
||
// 一万票的投票在有人投票期间每次刷新都要扫一万行 JSON，十万票直接把单请求拖到秒级。
|
||
//
|
||
// 现在改成：缓存过期后**先把旧值返回给调用方**，再在后台按投票去重地重算。
|
||
// 请求路径永远是 O(1)，O(N) 被挪到后台，且同一投票同时只有一个重算任务。
|
||
// 代价是结果最多陈旧 resultsCacheTTL（30s）—— 与原有缓存口径一致，结果页本就非强一致。
|
||
func LoadResults(ctx context.Context, v *db.Vote) (*Results, error) {
|
||
cacheKey := strconv.FormatUint(uint64(v.ID), 10)
|
||
|
||
if val, expiresAt, ok := resultsCache.GetWithExpiry(cacheKey); ok {
|
||
res, isResults := val.(*Results)
|
||
if !isResults {
|
||
// 类型不符理论上不会发生（key 由本文件的代码独占写入），
|
||
// 但裸断言一旦不符就是 panic —— 丢掉重新算即可。
|
||
resultsCache.Delete(cacheKey)
|
||
} else {
|
||
if time.Now().After(expiresAt) {
|
||
scheduleResultsRefresh(v, cacheKey)
|
||
}
|
||
return res, nil
|
||
}
|
||
}
|
||
|
||
return computeResultsAndCache(ctx.Request().Context(), v, cacheKey)
|
||
}
|
||
|
||
// scheduleResultsRefresh 在「当前投票没有刷新任务」时启动一次后台重算。
|
||
//
|
||
// 注意这里刻意不引用调用方传进来的 *db.Vote：handler 可能在之后继续改写它，
|
||
// 后台 goroutine 持有同一个指针就是潜在的数据竞争。只拷贝重算真正需要的三个字段。
|
||
func scheduleResultsRefresh(v *db.Vote, cacheKey string) {
|
||
if _, loaded := resultsRefreshing.LoadOrStore(cacheKey, struct{}{}); loaded {
|
||
return // 已有任务在跑，避免并发重算把 CPU 打满
|
||
}
|
||
|
||
voteID := v.ID
|
||
voteType := v.Type
|
||
allowCustom := v.AllowCustom
|
||
|
||
shared.SafeGo(func() {
|
||
defer resultsRefreshing.Delete(cacheKey)
|
||
|
||
reqCtx, cancel := stdctx.WithTimeout(stdctx.Background(), resultsRefreshTimeout)
|
||
defer cancel()
|
||
|
||
probe := &db.Vote{}
|
||
probe.ID = voteID
|
||
probe.Type = voteType
|
||
probe.AllowCustom = allowCustom
|
||
|
||
res, err := computeResults(reqCtx, probe)
|
||
if err != nil {
|
||
logrus.WithContext(reqCtx).WithError(err).WithField("vote_id", voteID).
|
||
Warn("后台重算投票结果失败，继续沿用旧结果直到下次成功")
|
||
return
|
||
}
|
||
resultsCache.Set(cacheKey, res, resultsCacheTTL)
|
||
})
|
||
}
|
||
|
||
// computeResultsAndCache 同步重算并写回缓存（仅用于缓存完全无值的冷启动）。
|
||
func computeResultsAndCache(reqCtx stdctx.Context, v *db.Vote, cacheKey string) (*Results, error) {
|
||
res, err := computeResults(reqCtx, v)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
resultsCache.Set(cacheKey, res, resultsCacheTTL)
|
||
return res, nil
|
||
}
|
||
|
||
// computeResults 按投票类型聚合统计（原来的 LoadResults 函数体）。
|
||
//
|
||
// 只依赖标准库 context：后台刷新时没有 flamego 的 Context 可用。
|
||
func computeResults(reqCtx stdctx.Context, v *db.Vote) (*Results, error) {
|
||
res := &Results{Type: v.Type}
|
||
total, err := db.VoteSubmissions.CountByVoteID(reqCtx, v.ID)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "count submissions")
|
||
}
|
||
res.Total = total
|
||
|
||
switch v.Type {
|
||
case db.VoteTypeSingle:
|
||
options, err := db.VoteOptions.ListByVoteID(reqCtx, v.ID)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote options")
|
||
}
|
||
// 一次分组查询替代逐选项 COUNT（避免 N+1）
|
||
counts, err := db.VoteSubmissions.CountGroupByOptionID(reqCtx, v.ID)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "count options")
|
||
}
|
||
for _, o := range options {
|
||
count := counts[o.ID]
|
||
percent := 0
|
||
if total > 0 {
|
||
percent = int(count * 100 / total)
|
||
}
|
||
res.Options = append(res.Options, &OptionStat{Option: o, Count: count, Percent: percent})
|
||
}
|
||
|
||
case db.VoteTypeMultiple:
|
||
options, err := db.VoteOptions.ListByVoteID(reqCtx, v.ID)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote options")
|
||
}
|
||
submissions, err := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote submissions")
|
||
}
|
||
counts := make(map[uint]int64, len(options))
|
||
var totalSelections int64
|
||
for _, s := range submissions {
|
||
var data struct {
|
||
Options []uint `json:"options"`
|
||
}
|
||
_ = json.Unmarshal(s.FormData, &data)
|
||
seen := make(map[uint]struct{}, len(data.Options))
|
||
for _, id := range data.Options {
|
||
if _, exists := seen[id]; exists {
|
||
continue
|
||
}
|
||
seen[id] = struct{}{}
|
||
counts[id]++
|
||
totalSelections++
|
||
}
|
||
}
|
||
for _, o := range options {
|
||
count := counts[o.ID]
|
||
percent := 0
|
||
// 多选：百分比分母取"总勾选数"（各项相加≈100%，符合"票数分布"直觉）；
|
||
// res.Total 仍是参与人数，用于"总参与"展示。
|
||
if totalSelections > 0 {
|
||
percent = int(count * 100 / totalSelections)
|
||
}
|
||
res.Options = append(res.Options, &OptionStat{Option: o, Count: count, Percent: percent})
|
||
}
|
||
|
||
case db.VoteTypeRating:
|
||
submissions, err := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote submissions")
|
||
}
|
||
scoreCounts := make(map[int]int64)
|
||
var sum int64
|
||
for _, s := range submissions {
|
||
var data struct {
|
||
Rating int `json:"rating"`
|
||
}
|
||
_ = json.Unmarshal(s.FormData, &data)
|
||
if data.Rating >= 1 && data.Rating <= 10 {
|
||
scoreCounts[data.Rating]++
|
||
sum += int64(data.Rating)
|
||
}
|
||
}
|
||
for score := 1; score <= 10; score++ {
|
||
res.Ratings = append(res.Ratings, &RatingStat{Score: score, Count: scoreCounts[score]})
|
||
}
|
||
if total > 0 {
|
||
res.AvgRating = float64(sum) / float64(total)
|
||
}
|
||
|
||
case db.VoteTypeRank:
|
||
options, err := db.VoteOptions.ListByVoteID(reqCtx, v.ID)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote options")
|
||
}
|
||
submissions, err := db.VoteSubmissions.ListByVoteID(reqCtx, v.ID, nil)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote submissions")
|
||
}
|
||
rankSums := make(map[uint]float64, len(options))
|
||
rankCounts := make(map[uint]int64, len(options))
|
||
for _, s := range submissions {
|
||
var data struct {
|
||
Ranking map[string]float64 `json:"ranking"`
|
||
}
|
||
_ = json.Unmarshal(s.FormData, &data)
|
||
for idStr, r := range data.Ranking {
|
||
id, _ := strconv.ParseUint(idStr, 10, 64)
|
||
rankSums[uint(id)] += r
|
||
rankCounts[uint(id)]++
|
||
}
|
||
}
|
||
for _, o := range options {
|
||
var avg float64
|
||
if rankCounts[o.ID] > 0 {
|
||
avg = rankSums[o.ID] / float64(rankCounts[o.ID])
|
||
}
|
||
res.Rankings = append(res.Rankings, &RankStat{Option: o, AvgRank: avg})
|
||
}
|
||
var rankSum float64
|
||
for _, r := range res.Rankings {
|
||
rankSum += r.AvgRank
|
||
}
|
||
if len(res.Rankings) > 0 {
|
||
res.AvgRank = rankSum / float64(len(res.Rankings))
|
||
}
|
||
|
||
case db.VoteTypeForm:
|
||
fields, err := db.VoteFormFields.ListByVoteID(reqCtx, v.ID)
|
||
if err != nil {
|
||
return nil, errors.Wrap(err, "list vote form fields")
|
||
}
|
||
res.FormFields = fields
|
||
res.FormCount = total
|
||
res.FormFieldStats = aggregateFormFieldStats(reqCtx, res.FormFields)
|
||
}
|
||
|
||
// 自定义补充选项聚合（single/multiple/rank 且开启 AllowCustom；评分/表单不适用）
|
||
if v.AllowCustom && (v.Type == db.VoteTypeSingle || v.Type == db.VoteTypeMultiple || v.Type == db.VoteTypeRank) {
|
||
res.CustomStats = aggregateCustomStats(reqCtx, v)
|
||
}
|
||
|
||
return res, nil
|
||
}
|