tamabox/internal/db/analytics.go

// Copyright 2022 E99p1ant. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package db

import (
	"context"
	"fmt"
	"time"

	"github.com/pkg/errors"
	"gorm.io/gorm"
)

// SiteAnalyticsStore 提供「全站公开数据页」所需的聚合查询。
//
// 与 internal/stats 的分工：本层只做数据库聚合（纯查询，不缓存、不生成文案），
// 缓存、补齐空日期与展示模型都在 stats 层。这样聚合 SQL 可以脱离 HTTP 单测。
type SiteAnalyticsStore interface {
	SiteAnalytics(ctx context.Context, now time.Time) (*SiteAnalytics, error)
	// TodayCounts 单独取「今日提问 / 今日回答」。
	//
	// 它们刻意不进 SiteAnalytics：整页聚合一整天只算一次，而这两个数要跟着
	// 页面上的「实时」说明走，必须每次（或每隔几十秒）重新取。
	TodayCounts(ctx context.Context, since time.Time) (questions, answers int64, err error)
}

// Analytics 全站聚合查询的包级入口，在 Init 中随其他 store 一起初始化。
var Analytics SiteAnalyticsStore

// SiteAnalytics 是一次全站聚合的原始结果，所有计数都是「累及至今」或「窗口内」的口径，
// 具体窗口由调用方传入的 now 决定（见 stats 包的 Overview）。
type SiteAnalytics struct {
	RegisteredUsers int64
	TotalQuestions  int64
	Answered        int64
	Private         int64
	Read            int64
	Anonymous       int64
	ActiveBoxes     int64 // 至少收到过一条提问的箱主数
	AnsweringBoxes  int64 // 至少回答过一条提问的箱主数
	TotalMessages   int64
	LastQuestionAt  *time.Time
	LastAnswerAt    *time.Time

	Daily   []CountPoint // 近 30 天（按日）
	Monthly []CountPoint // 近 12 个月（按月）

	// Heatmap 一周 7 天 × 24 小时的提问量，下标 0 为周日（与 EXTRACT(DOW) 对齐）。
	Heatmap     [7][24]int64
	HeatmapDays int // Heatmap 的统计窗口天数，用于在页面上说明口径

	LengthBins [6]int64 // 提问字数分布
	ReplyBins  [6]int64 // 回答耗时分布（仅统计已回答的提问）
}

// CountPoint 一个分组键上的计数，Key 已由数据库格式化成可直接展示的字符串。
type CountPoint struct {
	Key string
	N   int64
}

func NewSiteAnalyticsStore(db *gorm.DB) SiteAnalyticsStore {
	return &siteAnalytics{db}
}

type siteAnalytics struct {
	*gorm.DB
}

const (
	// 趋势窗口：近 30 天按日、近 12 个月按月。
	analyticsDailyDays = 30
	analyticsMonths    = 12
	// 热力图窗口：一年太贵（全表聚合），半年足以代表「全站习惯」。
	analyticsHeatmapDays = 180

	// 字数分桶上限（左闭右开），最后一桶为 500 以上。
	lengthBinExpr = `CASE
		WHEN CHAR_LENGTH(content) < 20 THEN 0
		WHEN CHAR_LENGTH(content) < 50 THEN 1
		WHEN CHAR_LENGTH(content) < 100 THEN 2
		WHEN CHAR_LENGTH(content) < 200 THEN 3
		WHEN CHAR_LENGTH(content) < 500 THEN 4
		ELSE 5 END`
)

// analyticsDialect 把两套数据库里含义相同、写法不同的表达式收敛到一处。
//
// 项目实际部署用 PostgreSQL（见 Init 的索引分支），MySQL 分支只是保留可用性：
// 配置切到 MySQL 时不应整页 500，而应安静地少几个数。
type analyticsDialect struct {
	day      string
	month    string
	dow      string
	hour     string
	replyMin string
}

func analyticsDialectFor(db *gorm.DB, at time.Time) analyticsDialect {
	tz := analyticsTZOffset(at)

	if db.Dialector.Name() == "mysql" {
		// MySQL 的 DATETIME 不带时区，驱动按 DSN 的 loc（默认 UTC）写入，
		// 因此库里存的是 UTC 墙钟，CONVERT_TZ 再搬到进程时区。用偏移而不是
		// 时区名：CONVERT_TZ 用时区名要求 mysql.time_zone* 表已导入。
		conv := func(col string) string {
			return fmt.Sprintf("CONVERT_TZ(%s, '+00:00', '%s')", col, tz)
		}
		day := conv("created_at")
		return analyticsDialect{
			day:      fmt.Sprintf("DATE_FORMAT(%s, '%%Y-%%m-%%d')", day),
			month:    fmt.Sprintf("DATE_FORMAT(%s, '%%Y-%%m')", day),
			dow:      fmt.Sprintf("(DAYOFWEEK(%s) - 1)", day),
			hour:     fmt.Sprintf("HOUR(%s)", day),
			replyMin: "TIMESTAMPDIFF(MINUTE, created_at, updated_at)",
		}
	}
	// PostgreSQL：EXTRACT 返回 numeric，显式转 int 以免扫描进 int64 时类型不匹配。
	loc := func(col string) string { return analyticsLocalize(col, at) }
	return analyticsDialect{
		day:      fmt.Sprintf("to_char(%s, 'YYYY-MM-DD')", loc("created_at")),
		month:    fmt.Sprintf("to_char(%s, 'YYYY-MM')", loc("created_at")),
		dow:      fmt.Sprintf("CAST(EXTRACT(DOW FROM %s) AS INTEGER)", loc("created_at")),
		hour:     fmt.Sprintf("CAST(EXTRACT(HOUR FROM %s) AS INTEGER)", loc("created_at")),
		replyMin: "CAST(EXTRACT(EPOCH FROM (updated_at - created_at)) / 60 AS INTEGER)",
	}
}

// analyticsLocalize 把带时区的列转成「Go 进程所在时区的墙钟时间」再参与日期分组。
//
// 不能依赖数据库会话时区：PG 官方镜像默认 timezone=UTC，而 Go 侧按容器本地时区
// （Dockerfile 里写的是 Asia/Shanghai）切「今天」。两边不一致时，日报表与热力图会
// 整体偏 8 小时 —— 中国时间 0–8 点的提问被算进前一天，且和页面顶部「今日」对不上。
// 这类偏差肉眼看不出来（数字本身都合理），只能靠显式换算消除。
//
// 用固定偏移而不是 IANA 名：Go 在容器未设 TZ 时会给出 "Local"，PG 不认这个名字。
// 中国无夏令时，固定偏移与真实时区完全等价；有夏令时的部署在窗口内可能差 1 小时。
func analyticsLocalize(col string, at time.Time) string {
	return fmt.Sprintf("%s AT TIME ZONE INTERVAL '%s'", col, analyticsTZOffset(at))
}

// analyticsTZOffset 返回 +08:00 这样的偏移字面量。
func analyticsTZOffset(at time.Time) string {
	_, off := at.Zone()
	sign := "+"
	if off < 0 {
		sign, off = "-", -off
	}
	return fmt.Sprintf("%s%02d:%02d", sign, off/3600, off%3600/60)
}

func (db *siteAnalytics) SiteAnalytics(ctx context.Context, now time.Time) (*SiteAnalytics, error) {
	loc := now.Location()
	startOfDay := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
	dailySince := startOfDay.AddDate(0, 0, -(analyticsDailyDays - 1))
	// 先归到当月 1 号再回退，避免月底 AddDate 溢出（3-31 减 11 个月会滚到 3-03）。
	monthSince := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, loc).AddDate(0, -(analyticsMonths - 1), 0)
	heatmapSince := startOfDay.AddDate(0, 0, -analyticsHeatmapDays)

	out := &SiteAnalytics{HeatmapDays: analyticsHeatmapDays}

	if err := db.aggregateTotals(ctx, out); err != nil {
		return nil, err
	}
	if err := db.aggregateTrends(ctx, out, now, dailySince, monthSince); err != nil {
		return nil, err
	}
	if err := db.aggregateHeatmap(ctx, out, now, heatmapSince); err != nil {
		return nil, err
	}
	if err := db.aggregateContent(ctx, out); err != nil {
		return nil, err
	}

	if err := db.WithContext(ctx).Model(&User{}).Count(&out.RegisteredUsers).Error; err != nil {
		return nil, errors.Wrap(err, "count registered users")
	}
	if err := db.WithContext(ctx).Model(&Message{}).Count(&out.TotalMessages).Error; err != nil {
		return nil, errors.Wrap(err, "count messages")
	}

	// 「今日」两项不在这里查：本函数的结果整天缓存，把会变的数字放进来
	// 会让页面顶着一个「实时」的说明显示一整天不变的值。见 TodayCounts。
	return out, nil
}

// TodayCounts 取今日提问数与今日回答数，只扫当天数据（走 created_at 索引）。
//
// 口径：回答数按「今天被回答过」计，用的是 updated_at —— 置顶等写操作也会刷新它，
// 因此这个数偏大。页面只把它当「今天有多少动静」的量级参考，不做精确承诺。
func (db *siteAnalytics) TodayCounts(ctx context.Context, since time.Time) (int64, int64, error) {
	var questions, answers int64
	if err := db.WithContext(ctx).Model(&Question{}).
		Where("created_at >= ?", since).Count(&questions).Error; err != nil {
		return 0, 0, errors.Wrap(err, "count today's questions")
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Where("answer <> '' AND updated_at >= ?", since).Count(&answers).Error; err != nil {
		return 0, 0, errors.Wrap(err, "count today's answers")
	}
	return questions, answers, nil
}

// aggregateTotals 用一条 SQL 取回所有「不分时间窗口」的计数。
//
// 用 SUM(CASE WHEN ...) 而不是 COUNT(*) FILTER (WHERE ...)：前者 MySQL 也支持，
// 后者只在 PostgreSQL 上可用。
func (db *siteAnalytics) aggregateTotals(ctx context.Context, out *SiteAnalytics) error {
	var row struct {
		Total          int64
		Answered       int64
		Private        int64
		Read           int64
		Anonymous      int64
		ActiveBoxes    int64      `gorm:"column:active_boxes"`
		AnsweringBoxes int64      `gorm:"column:answering_boxes"`
		LastQ          *time.Time `gorm:"column:last_q"`
		LastA          *time.Time `gorm:"column:last_a"`
	}
	err := db.WithContext(ctx).Model(&Question{}).
		Select(`COUNT(*) AS total,
			COALESCE(SUM(CASE WHEN answer <> '' THEN 1 ELSE 0 END), 0) AS answered,
			COALESCE(SUM(CASE WHEN is_private THEN 1 ELSE 0 END), 0) AS private,
			COALESCE(SUM(CASE WHEN is_read OR answer <> '' THEN 1 ELSE 0 END), 0) AS read,
			COALESCE(SUM(CASE WHEN asker_user_id = 0 THEN 1 ELSE 0 END), 0) AS anonymous,
			COUNT(DISTINCT user_id) AS active_boxes,
			COUNT(DISTINCT CASE WHEN answer <> '' THEN user_id END) AS answering_boxes,
			MAX(created_at) AS last_q,
			MAX(CASE WHEN answer <> '' THEN updated_at END) AS last_a`).
		Scan(&row).Error
	if err != nil {
		return errors.Wrap(err, "aggregate question totals")
	}

	out.TotalQuestions = row.Total
	out.Answered = row.Answered
	out.Private = row.Private
	out.Read = row.Read
	out.Anonymous = row.Anonymous
	out.ActiveBoxes = row.ActiveBoxes
	out.AnsweringBoxes = row.AnsweringBoxes
	out.LastQuestionAt = row.LastQ
	out.LastAnswerAt = row.LastA
	return nil
}

// aggregateTrends 按日与按月各取一次计数。未被 GROUP BY 覆盖的空日期由 stats 层补齐。
func (db *siteAnalytics) aggregateTrends(ctx context.Context, out *SiteAnalytics, now, dailySince, monthSince time.Time) error {
	d := analyticsDialectFor(db.DB, now)

	var daily []struct {
		Key string
		N   int64
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Select(d.day+" AS key, COUNT(*) AS n").
		Where("created_at >= ?", dailySince).
		Group(d.day).Order(d.day).
		Scan(&daily).Error; err != nil {
		return errors.Wrap(err, "aggregate daily trend")
	}
	for _, row := range daily {
		out.Daily = append(out.Daily, CountPoint{Key: row.Key, N: row.N})
	}

	var monthly []struct {
		Key string
		N   int64
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Select(d.month+" AS key, COUNT(*) AS n").
		Where("created_at >= ?", monthSince).
		Group(d.month).Order(d.month).
		Scan(&monthly).Error; err != nil {
		return errors.Wrap(err, "aggregate monthly trend")
	}
	for _, row := range monthly {
		out.Monthly = append(out.Monthly, CountPoint{Key: row.Key, N: row.N})
	}
	return nil
}

func (db *siteAnalytics) aggregateHeatmap(ctx context.Context, out *SiteAnalytics, now, since time.Time) error {
	d := analyticsDialectFor(db.DB, now)

	var rows []struct {
		Dow int
		Hr  int
		N   int64
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Select(d.dow+" AS dow, "+d.hour+" AS hr, COUNT(*) AS n").
		Where("created_at >= ?", since).
		Group(d.dow + ", " + d.hour).
		Scan(&rows).Error; err != nil {
		return errors.Wrap(err, "aggregate weekday/hour heatmap")
	}

	for _, row := range rows {
		// 越界值直接丢弃：MySQL 的 DAYOFWEEK 或异常时区数据都可能给出 7 以上的下标，
		// 落进固定数组会 panic，而这份数据只影响一张图，不值得让整页失败。
		if row.Dow < 0 || row.Dow > 6 || row.Hr < 0 || row.Hr > 23 {
			continue
		}
		out.Heatmap[row.Dow][row.Hr] = row.N
	}
	return nil
}

func (db *siteAnalytics) aggregateContent(ctx context.Context, out *SiteAnalytics) error {
	// 内容分布只用时长差值，不涉及日期分组，时区不影响结果。
	d := analyticsDialectFor(db.DB, time.Now())

	var lengths []struct {
		Bin int
		N   int64
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Select(lengthBinExpr + " AS bin, COUNT(*) AS n").
		Group(lengthBinExpr).
		Scan(&lengths).Error; err != nil {
		return errors.Wrap(err, "aggregate content length distribution")
	}
	for _, row := range lengths {
		if row.Bin >= 0 && row.Bin < len(out.LengthBins) {
			out.LengthBins[row.Bin] = row.N
		}
	}

	// 回答耗时：只用已回答的提问，按分钟分桶。
	// 口径注意：updated_at 也会被置顶等写操作刷新，因此这里算出的耗时偏大；
	// 展示时只呈现「24 小时内回答占比」这类粗粒度结论，不宣称精确中位数。
	replyBinExpr := fmt.Sprintf(`CASE
		WHEN %[1]s < 60 THEN 0
		WHEN %[1]s < 360 THEN 1
		WHEN %[1]s < 1440 THEN 2
		WHEN %[1]s < 4320 THEN 3
		WHEN %[1]s < 10080 THEN 4
		ELSE 5 END`, d.replyMin)

	var replies []struct {
		Bin int
		N   int64
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Select(replyBinExpr + " AS bin, COUNT(*) AS n").
		Where("answer <> ''").
		Group(replyBinExpr).
		Scan(&replies).Error; err != nil {
		return errors.Wrap(err, "aggregate reply latency distribution")
	}
	for _, row := range replies {
		if row.Bin >= 0 && row.Bin < len(out.ReplyBins) {
			out.ReplyBins[row.Bin] = row.N
		}
	}
	return nil
}