数据分析与统计：埋点体系、事件模型与内容热度计算

// Go 埋点接收服务：只做接收与入队，不做任何重逻辑
package tracker

import (
	"encoding/json"
	"net/http"
	"time"

	"github.com/segmentio/kafka-go"
)

type TrackEvent struct {
	Event     string         `json:"event"`                // 事件名，如 post_create
	UserID    int64          `json:"user_id"`
	SessionID string         `json:"session_id"`
	Ts        int64          `json:"ts"`
	Props     map[string]any `json:"props"`                // 业务属性
	Device    struct {
		Platform string `json:"platform"` // web | ios | android
		OS       string `json:"os"`
		Browser  string `json:"browser"`
	} `json:"device"`
}

func (s *Server) handleTrack(w http.ResponseWriter, r *http.Request) {
	var ev TrackEvent
	if err := json.NewDecoder(r.Body).Decode(&ev); err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	// 规范化：补充服务端时间，防止客户端时钟偏差
	ev.Ts = time.Now().UnixMilli()

	// 异步写入 Kafka，接口立即返回 204
	msg := kafka.Message{Topic: "tracking-events", Value: mustJSON(ev)}
	select {
	case s.producerCh <- msg:
		w.WriteHeader(http.StatusNoContent)
	default:
		// 队列满时降级：直接丢弃，宁可丢埋点也不阻塞业务
		w.WriteHeader(http.StatusNoContent)
	}
}

-- 事件表 (原始行为流水)
CREATE TABLE tracking_events (
    event_id  BIGINT,
    event     String,      -- post_view / post_like / post_repost
    user_id   UInt64,
    post_id   UInt64,
    ts        DateTime64(3),
    props     String       -- JSON 扩展属性
) ENGINE = MergeTree
PARTITION BY toYYYYMMDD(ts)
ORDER BY (event, post_id, ts);

-- 事实表 (内容状态快照，周期性聚合)
CREATE TABLE post_facts (
    post_id       UInt64,
    view_count    UInt64,
    like_count    UInt64,
    repost_count  UInt64,
    reply_count   UInt64,
    heat_score    Float64,
    snapshot_date Date
) ENGINE = SummingMergeTree
PARTITION BY toYYYYMMDD(snapshot_date)
ORDER BY post_id;

-- 基于事件流的漏斗查询：计算每环节的用户数
WITH funnel AS (
    SELECT
        user_id,
        max(step) AS max_step
    FROM (
        SELECT
            user_id,
            CASE
                WHEN event = 'register'           THEN 1
                WHEN event = 'profile_complete'   THEN 2
                WHEN event = 'timeline_view'      THEN 3
                WHEN event = 'post_create'        THEN 4
                WHEN event = 'post_published'     THEN 5
                ELSE 0
            END AS step
        FROM tracking_events
        WHERE ts >= toDateTime('2026-09-01') AND ts < toDateTime('2026-09-08')
    )
    GROUP BY user_id
)
SELECT
    sum(max_step >= 1) AS register_users,
    sum(max_step >= 2) AS profile_users,
    sum(max_step >= 3) AS timeline_users,
    sum(max_step >= 4) AS create_users,
    sum(max_step >= 5) AS published_users
FROM funnel;

import math

def hn_score(votes: int, created_at: float, gravity: float = 1.8) -> float:
    """Hacker News 风格热度分：分母的幂次控制衰减速度"""
    age_hours = (now() - created_at) / 3600.0
    return (votes - 1) / (age_hours + 2) ** gravity

def heat_score(post) -> float:
    """微型博客热度分：多信号加权 + 对数衰减"""
    age_hours = (now() - post.created_at) / 3600.0

    # 1. 互动加权分（点赞/转发/评论权重递增，评论代表深度互动）
    like_w   = post.like_count   * 1.0
    repost_w = post.repost_count * 3.0   # 转发=二次传播，权重最高
    reply_w  = post.reply_count  * 2.0
    view_w   = post.view_count   * 0.05

    interaction = like_w + repost_w + reply_w + view_w

    # 2. 作者质量归一化（防止大 V 刷榜，除以粉丝数的对数）
    author_norm = math.log10(post.author_followers + 10)

    # 3. 分段衰减：前 6 小时陡峭，之后平缓
    if age_hours < 6:
        decay = 1.0
    elif age_hours < 24:
        decay = 1 / (1 + 0.2 * (age_hours - 6))
    else:
        decay = 1 / (1 + 0.05 * age_hours)

    return (interaction / author_norm) * decay

// Go 定时任务：每小时重算全站热度 Top 内容
func (s *HeatService) RecalculateHeat(ctx context.Context) error {
	// 1. 从数仓拉取最近 72h 的互动聚合
	agg := s.warehouse.QueryRecentInteractions(ctx, time.Hour*72)

	// 2. 批量计算热度分
	scoreMap := make(map[int64]float64, len(agg))
	for _, post := range agg {
		scoreMap[post.ID] = computeHeatScore(post)
	}

	// 3. 写回 Redis（榜单读取直接命中）
	pipe := s.redis.Pipeline()
	for id, score := range scoreMap {
		pipe.ZAdd(ctx, "hot_ranking", redis.Z{Score: score, Member: id})
	}
	_, err := pipe.Exec(ctx)
	return err
}

-- ClickHouse 物化视图：每分钟实时聚合在线活跃数
CREATE MATERIALIZED VIEW mv_online_users
ENGINE = SummingMergeTree
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (timestamp, platform)
AS
SELECT
    toStartOfMinute(ts) AS timestamp,
    platform,
    uniqExact(user_id)  AS active_users
FROM tracking_events
GROUP BY timestamp, platform;