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

// 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 // 回答耗时分布（仅统计已回答的提问）

	TodayQuestions int64
	TodayAnswers   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) analyticsDialect {
	if db.Dialector.Name() == "mysql" {
		return analyticsDialect{
			day:      "DATE_FORMAT(created_at, '%Y-%m-%d')",
			month:    "DATE_FORMAT(created_at, '%Y-%m')",
			dow:      "(DAYOFWEEK(created_at) - 1)",
			hour:     "HOUR(created_at)",
			replyMin: "TIMESTAMPDIFF(MINUTE, created_at, updated_at)",
		}
	}
	// PostgreSQL：EXTRACT 返回 numeric，显式转 int 以免扫描进 int64 时类型不匹配。
	return analyticsDialect{
		day:      "to_char(created_at, 'YYYY-MM-DD')",
		month:    "to_char(created_at, 'YYYY-MM')",
		dow:      "CAST(EXTRACT(DOW FROM created_at) AS INTEGER)",
		hour:     "CAST(EXTRACT(HOUR FROM created_at) AS INTEGER)",
		replyMin: "CAST(EXTRACT(EPOCH FROM (updated_at - created_at)) / 60 AS INTEGER)",
	}
}

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, dailySince, monthSince); err != nil {
		return nil, err
	}
	if err := db.aggregateHeatmap(ctx, out, 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")
	}

	// 今日数字单独查：它们不进「每天一次」的缓存窗口，页面顶部要显示实时值。
	if err := db.WithContext(ctx).Model(&Question{}).
		Where("created_at >= ?", startOfDay).Count(&out.TodayQuestions).Error; err != nil {
		return nil, errors.Wrap(err, "count today's questions")
	}
	if err := db.WithContext(ctx).Model(&Question{}).
		Where("answer <> '' AND updated_at >= ?", startOfDay).Count(&out.TodayAnswers).Error; err != nil {
		return nil, errors.Wrap(err, "count today's answers")
	}

	return out, 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, dailySince, monthSince time.Time) error {
	d := analyticsDialectFor(db.DB)

	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, since time.Time) error {
	d := analyticsDialectFor(db.DB)

	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)

	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
}